标签：
package com.shma.util;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
public class ObjectUtil {
/**对象转byte[]
* @param obj
* @return
* @throws IOException
*/
public static byte[] objectToBytes(Object obj) throws Exception{
ByteArrayOutputStream bo = new ByteArrayOutputStream();
ObjectOutputStream oo = new ObjectOutputStream(bo);
oo.writeObject(obj);
byte[] bytes = bo.toByteArray();
bo.close();
oo.close();
return bytes;
}
/**byte[]转对象
* @param bytes
* @return
* @throws Exception
*/
public static Object bytesToObject(byte[] bytes) throws Exception{
ByteArrayInputStream in = new ByteArrayInputStream(bytes);
ObjectInputStream sIn = new ObjectInputStream(in);
return sIn.readObject();
}
}
package com.shma.util;
import java.util.Random;
public class RedPacketAlgorithm {
static Random random = new Random();
static {
random.setSeed(System.currentTimeMillis());
}
public static void main(String[] args) {
long max = 20;
long min = 1;
long[] result = generate(100, 10, max, min);
long total = 0;
for (int i = 0; i < result.length; i++) {
total += result[i];
}
//检查生成的红包的总额是否正确
System.out.println("total:" + total);
//统计每个钱数的红包数量，检查是否接近正态分布
int count[] = new int[(int) max + 1];
for (int i = 0; i < result.length; i++) {
count[(int) result[i]] += 1;
}
for (int i = 0; i < count.length; i++) {
System.out.println("" + i + " " + count[i]);
}
}
/**
* 生产min和max之间的随机数，但是概率不是平均的，从min到max方向概率逐渐加大。
* 先平方，然后产生一个平方值范围内的随机数，再开方，这样就产生了一种“膨胀”再“收缩”的效果。
*
* @param min
* @param max
* @return
*/
static long xRandom(long min, long max) {
return sqrt(nextLong(sqr(max - min)));
}
/**
*
* @param total
* 红包总额
* @param count
* 红包个数
* @param max
* 每个小红包的最大额
* @param min
* 每个小红包的最小额
* @return 存放生成的每个小红包的值的数组
*/
public static long[] generate(long total, int count, long max, long min) {
if(count * max < total) {
System.out.println("最大红包钱数 * 红包个数 < 总钱数");
System.exit(-1);
}
long[] result = new long[count];
long average = total / count;
long a = average - min;
long b = max - min;
//
//这样的随机数的概率实际改变了，产生大数的可能性要比产生小数的概率要小。
//这样就实现了大部分红包的值在平均数附近。大红包和小红包比较少。
long range1 = sqr(average - min);
long range2 = sqr(max - average);
for (int i = 0; i < result.length; i++) {
//因为小红包的数量通常是要比大红包的数量要多的，因为这里的概率要调换过来。
//当随机数>平均值，则产生小红包
//当随机数<平均值，则产生大红包
if (nextLong(min, max) > average) {
// 在平均线上减钱
// long temp = min + sqrt(nextLong(range1));
long temp = min + xRandom(min, average);
result[i] = temp;
total -= temp;
} else {
// 在平均线上加钱
// long temp = max - sqrt(nextLong(range2));
long temp = max - xRandom(average, max);
result[i] = temp;
total -= temp;
}
}
// 如果还有余钱，则尝试加到小红包里，如果加不进去，则尝试下一个。
while (total > 0) {
for (int i = 0; i < result.length; i++) {
if (total > 0 && result[i] < max) {
result[i]++;
total--;
}
}
}
// 如果钱是负数了，还得从已生成的小红包中抽取回来
while (total < 0) {
for (int i = 0; i < result.length; i++) {
if (total < 0 && result[i] > min) {
result[i]--;
total++;
}
}
}
return result;
}
static long sqrt(long n) {
// 改进为查表？
return (long) Math.sqrt(n);
}
static long sqr(long n) {
// 查表快，还是直接算快？
return n * n;
}
static long nextLong(long n) {
return random.nextInt((int) n);
}
static long nextLong(long min, long max) {
return random.nextInt((int) (max - min + 1)) + min;
}
}
package com.shma;
import java.io.Serializable;
public class Msg implements Serializable {
private static final long serialVersionUID = -948478514538813354L;
private int id;
private long money;
private long sourceUserId;
private long getUserId;
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public long getMoney() {
return money;
}
public void setMoney(long money) {
this.money = money;
}
public long getSourceUserId() {
return sourceUserId;
}
public void setSourceUserId(long sourceUserId) {
this.sourceUserId = sourceUserId;
}
public long getGetUserId() {
return getUserId;
}
public void setGetUserId(long getUserId) {
this.getUserId = getUserId;
}
@Override
public String toString() {
return "Msg [id=" + id + ", money=" + money + ", sourceUserId="
+ sourceUserId + ", getUserId=" + getUserId + "]";
}
}
package test;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import org.apache.log4j.Logger;
import org.junit.Before;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.util.StopWatch;
import com.shma.Msg;
import com.shma.redis.RedisUtil;
import com.shma.redis.RedisUtil.RedisHash;
import com.shma.util.ObjectUtil;
import com.shma.util.RedPacketAlgorithm;
public class TestRedis {
protected final static Logger logger = Logger.getLogger("executeLog");
private ApplicationContext context;
@Before
public void init() throws Exception {
String path = "/config/spring-application.xml";
logger.info(path);
context = new ClassPathXmlApplicationContext(path);
}
@Test
public void test() {
final RedisUtil redisUtil = context.getBean(RedisUtil.class);
long userId = 45243043L;
int max = 100;
int min = 2;
int total = 10000;
int count = 200;
//将大红包拆分成小红包数组
long[] rpDatas = RedPacketAlgorithm.generate(total, count, max, min);
// 将生成的红包放入队列中
final String key = "red_packet_original_queue9_" + userId;
for (int i = 0; i < rpDatas.length; ++i) {
Msg msg = new Msg();
msg.setId(i + 1);
msg.setMoney(rpDatas[i]);
msg.setSourceUserId(userId);
logger.info("key : " + key + ", msg : " + msg);
try {
redisUtil.lpush(key.getBytes(), ObjectUtil.objectToBytes(msg));
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
StopWatch watch = new StopWatch();
logger.info("start:" + System.currentTimeMillis()/1000);
// 已经领取用户列表
final String mapKey = "map9_" + key;
// 最后生成的领取记录队列
final String finshKey = "finish9_" + key;
/**
* 接收到一个获取红包请求 判断是否已经领取过
* 如果没有，则从redis队列中取出一个分给该用户
* 添加该用户的领取记录
* 如果没有，则返回已经领取完 将完成队列处理入库
*/
//模拟300人同时不同的抢红包
int threadNum = 300;
final CountDownLatch latch = new CountDownLatch(threadNum);
watch.start();
for (int i = 0; i < threadNum; ++i) {
final int temp = i;
Thread thread = new Thread() {
public void run() {
try {
String lockKey = key + "_" + temp;
while(true) {
//加锁60秒
if(redisUtil.isExistUpdate(lockKey, "60")) {
RedisHash redisHash = redisUtil.getRedisHash(mapKey);
if(!redisHash.isExist(""+temp)) {
redisHash.setOnlyIfNotExists(temp+"", "1");
try {
logger.info("size:" + redisUtil.getLen(key.getBytes()));
byte[] data = redisUtil.lpop(key.getBytes());
if(data != null && data.length > 0) {
Msg msg = (Msg)ObjectUtil.bytesToObject(data);
msg.setGetUserId(temp);
logger.info("userid:" + temp + ", msg:" + msg);
redisUtil.lpush(finshKey.getBytes(), ObjectUtil.objectToBytes(msg));
} else {
if(redisUtil.getLen(key.getBytes()) <= 0) {
logger.info("尊敬的[" + temp + "]用户，" + "已经领取完了");
break;
}
}
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
} else {
logger.info("尊敬的[" + temp + "]用户，您已经领取过了");
}
redisUtil.unLockRedisKey(lockKey);
}
if(redisUtil.getLen(key.getBytes()) <= 0) {
logger.info("尊敬的[" + temp + "]用户，" + "已经领取完了");
break;
}
}
} finally {
latch.countDown();
}
}
};
thread.start();
}
try {
latch.await();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
watch.stop();
logger.info(redisUtil.getRedisList(key.getBytes()));
logger.info(redisUtil.getRedisList(finshKey.getBytes()));
List<byte[]> msgs = redisUtil.getRedisList(finshKey.getBytes());
for(byte[] msg : msgs) {
Msg msgsMsg = null;
try {
msgsMsg = (Msg) ObjectUtil.bytesToObject(msg);
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
logger.info(msgsMsg);
}
logger.info("time:" + watch.getTotalTimeSeconds());
logger.info("speed:" + total/watch.getTotalTimeSeconds());
logger.info("end:" + System.currentTimeMillis()/1000);
}
}
package com.shma.redis;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import redis.clients.jedis.ShardedJedis;
public class RedisUtil {
private RedisService redisService;
public void set(String key, String value) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
shardedJedis.set(key, value);
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
}
public String get(String key) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
return shardedJedis.get(key);
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
return null;
}
public boolean del(String key) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
return shardedJedis.del(key) > 0 ? true : false;
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
return false;
}
/**
* 存储到redis队列中，插入到表头
*
* @param key
* @param value
*/
public void lpush(byte[] key, byte[] value) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
shardedJedis.lpush(key, value);
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
}
/**
* 存储到redis队列中，插入到表尾
*
* @param key
* @param value
*/
public void rpush(byte[] key, byte[] value) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
shardedJedis.rpush(key, value);
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
}
public byte[] lpop(byte[] key) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
return shardedJedis.lpop(key);
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
return null;
}
public byte[] rpop(byte[] key) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
return shardedJedis.rpop(key);
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
return null;
}
public Long getLen(byte[] key) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
return shardedJedis.llen(key);
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
return 0L;
}
public List<byte[]> getRedisList(byte[] key) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
return shardedJedis.lrange(key, 0, -1);
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
return null;
}
public boolean isExistUpdate(final String... param) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
String key = param[0];
int expire = 20;
if (param.length > 1) {
expire = Integer.parseInt(param[1]);
}
long status = shardedJedis.setnx("redis_lock_" + key, "true");
if (status > 0) {
shardedJedis.expire("redis_lock_" + key, expire);
}
return status <= 0 ? true : false;
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
return false;
}
public Long unLockRedisKey(final String key) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
return shardedJedis.del("redis_lock_" + key);
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
return -1L;
}
public RedisHash getRedisHash(String key) {
return new RedisHash(key);
}
/**
* Redis 哈希
*/
public class RedisHash {
private String key;
public RedisHash(String key) {
this.key = key;
}
/**
* 获取指定属性值
*
* @param field 属性名
*
* @return 属性值
*/
public String get(final String field) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
return shardedJedis.hget(key, field);
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
return null;
}
/**
* 获取指定属性值
*
* @param fields 属性名
*
* @return 属性值
*/
public List<String> get(final String... fields) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
return shardedJedis.hmget(key, fields);
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
return null;
}
/**
* 设置属性
*
* @param field 属性名
* @param value 属性值
*/
public void put(final String field, final String value) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
shardedJedis.hset(key, field, value);
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
}
/**
* 仅当属性名不存在是设置属性
*
* @param field 属性名
* @param value 属性值
*
* @return 0表示属性已存在
*/
public int setOnlyIfNotExists(final String field, final String value) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
return shardedJedis.hsetnx(key, field, value).intValue();
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
return -1;
}
/**
* 保存多个属性名和属性值
*
* @param map 属性
*/
public void putAll(final Map<String, String> map) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
shardedJedis.hmset(key, map);
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
}
/**
* 删除一个或多个属性
*
* @param fields 属性名
*
* @return 被删除的属性数量
*/
public int delete(final String... fields) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
return shardedJedis.hdel(key, fields).intValue();
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
return -1;
}
/**
* 列出所有属性
*
* @return 所有属性名
*/
public List<String> keys() {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
return new ArrayList<String>(shardedJedis.hkeys(key));
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
return null;
}
/**
* 读取所有属性值并转换为 Map 对象
*
* @return 所有属性值
*/
public Map<String, String> toMap() {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
return shardedJedis.hgetAll(key);
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
return null;
}
/**
* 读取key的长度
*
* @return 所有属性值
*/
public Long getLen() {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
return shardedJedis.hlen(key);
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
return 0L;
}
/**
* 是否存在一个key
*
* @return 所有属性值
*/
public Boolean isExist(final String field) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = redisService.getShareJedisPoolConnection();
return shardedJedis.hexists(key, field);
} catch (Throwable e) {
e.printStackTrace();
} finally {
shardedJedis.close();
}
return false;
}
}
public RedisService getRedisService() {
return redisService;
}
public void setRedisService(RedisService redisService) {
this.redisService = redisService;
}
}
package com.shma.redis;
import java.util.List;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import redis.clients.jedis.JedisShardInfo;
import redis.clients.jedis.ShardedJedis;
import redis.clients.jedis.ShardedJedisPool;
public class RedisService {
private GenericObjectPoolConfig jedisPoolConfig;
private List<JedisShardInfo> jedisShardInfos;
private ShardedJedisPool shareJedisPool;
public void init() {
shareJedisPool =new ShardedJedisPool(jedisPoolConfig, jedisShardInfos);
}
public ShardedJedis getShareJedisPoolConnection() {
ShardedJedis shardedJedis = shareJedisPool.getResource();
return shardedJedis;
}
public GenericObjectPoolConfig getJedisPoolConfig() {
return jedisPoolConfig;
}
public void setJedisPoolConfig(GenericObjectPoolConfig jedisPoolConfig) {
this.jedisPoolConfig = jedisPoolConfig;
}
public List<JedisShardInfo> getJedisShardInfos() {
return jedisShardInfos;
}
public void setJedisShardInfos(List<JedisShardInfo> jedisShardInfos) {
this.jedisShardInfos = jedisShardInfos;
}
public ShardedJedisPool getShareJedisPool() {
return shareJedisPool;
}
public void setShareJedisPool(ShardedJedisPool shareJedisPool) {
this.shareJedisPool = shareJedisPool;
}
}
Java-No.14 基于redis分布式缓存队列实现抢红包功能
标签：
原文地址：http://my.oschina.net/shma1664/blog/487218
