基于Redis实现分布式消息队列（4）
1、访问Redis的工具类
public class RedisManager {
private static Pool<Jedis> pool;
protected final static Logger logger = Logger.getLogger(RedisManager.class);
static{
try {
init();
} catch (Exception e) {
e.printStackTrace();
}
}
public static void init() throws Exception {
Properties props = ConfigManager.getProperties("redis");
logger.debug("初始化Redis连接池。");
if(props==null){
throw new RuntimeException("没有找到redis配置文件");
}
// 创建jedis池配置实例
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
// 设置池配置项值
int poolMaxTotal = Integer.valueOf(props.getProperty("redis.pool.maxTotal").trim());
jedisPoolConfig.setMaxTotal(poolMaxTotal);
int poolMaxIdle = Integer.valueOf(props.getProperty("redis.pool.maxIdle").trim());
jedisPoolConfig.setMaxIdle(poolMaxIdle);
long poolMaxWaitMillis = Long.valueOf(props.getProperty("redis.pool.maxWaitMillis").trim());
jedisPoolConfig.setMaxWaitMillis(poolMaxWaitMillis);
logger.debug(String.format("poolMaxTotal: %s , poolMaxIdle : %s , poolMaxWaitMillis : %s ",
poolMaxTotal,poolMaxIdle,poolMaxWaitMillis));
// 根据配置实例化jedis池
String connectMode = props.getProperty("redis.connectMode");
String hostPortStr = props.getProperty("redis.hostPort");
logger.debug(String.format("host : %s ",hostPortStr));
logger.debug(String.format("mode : %s ",connectMode));
if(StringUtils.isEmpty(hostPortStr)){
throw new OptimusException("redis配置文件未配置主机-端口集");
}
String[] hostPortSet = hostPortStr.split(",");
if("single".equals(connectMode)){
String[] hostPort = hostPortSet[0].split(":");
pool = new JedisPool(jedisPoolConfig, hostPort[0], Integer.valueOf(hostPort[1].trim()));
}else if("sentinel".equals(connectMode)){
Set<String> sentinels = new HashSet<String>();
for(String hostPort : hostPortSet){
sentinels.add(hostPort);
}
pool = new JedisSentinelPool("mymaster", sentinels, jedisPoolConfig);
}
}
/**
* 使用完成后，必须调用 returnResource 还回。
* @return 获取Jedis对象
*/
public static Jedis getResource(){
Jedis jedis = pool.getResource();
if(logger.isDebugEnabled()){
logger.debug("获得链接：" + jedis);
}
return jedis;
}
/**
* 获取Jedis对象。
*
* 用完后，需要调用returnResource放回连接池。
*
* @param db 数据库序号
* @return
*/
public static Jedis getResource(int db){
Jedis jedis = pool.getResource();
jedis.select(db);
if(logger.isDebugEnabled()){
logger.debug("获得链接：" + jedis);
}
return jedis;
}
/**
* @param jedis
*/
public static void returnResource(Jedis jedis){
if(jedis!=null){
pool.returnResource(jedis);
if(logger.isDebugEnabled()){
logger.debug("放回链接：" + jedis);
}
}
}
/**
* 需要通过Spring确认这个方法被调用。
* @throws Exception
*/
public static void destroy() throws Exception {
pool.destroy();
}
}
这个类没有通过技术手段强制调用returnResource和destroy，需要想想办法。
2、队列接口
public interface TaskQueue {
/**
* 获取队列名
* @return
*/
String getName();
/**
* 往队列中添加任务
* @param task
*/
void pushTask(String task);
/**
* 从队列中取出一个任务
* @return
*/
String popTask();
}
用String类型描述任务，也可以考虑byte[]，要求对每个任务描述的数据尽可能短。
3、队列的Redis实现类
/**
* 任务队列Redis实现。
*
* 采用每次获取Jedis并放回pool的方式。
* 如果获得Jedis后一直不放手，反复重用，两个操作耗时可以降低1/3。
* 暂时先忍受这种低性能，不明确Jedis是否线程安全。
*
*/
public class TaskQueueRedisImpl implements TaskQueue {
private final static int REDIS_DB_IDX = 9;
private final static Logger logger = Logger.getLogger(TaskQueueRedisImpl.class);
private final String name;
/**
* 构造函数。
*
* @param name
*/
public TaskQueueRedisImpl(String name) {
this.name = name;
}
/* (non-Javadoc)
* @see com.gwssi.common.mq.TaskQueue#getName()
*/
public String getName() {
return this.name;
}
/* (non-Javadoc)
* @see com.gwssi.common.mq.TaskQueue#pushTask(String)
*/
public void pushTask(String task) {
Jedis jedis = null;
try{
jedis = RedisManager.getResource(REDIS_DB_IDX);
jedis.lpush(this.name, task);
}catch(Throwable e){
logger.error(e.getMessage(),e);
}finally{
if(jedis!=null){
RedisManager.returnResource(jedis);
}
}
}
/* (non-Javadoc)
* @see com.gwssi.common.mq.TaskQueue#popTask()
*/
public String popTask() {
Jedis jedis = null;
String task = null;
try{
jedis = RedisManager.getResource(REDIS_DB_IDX);
task = jedis.rpop(this.name);
}catch(Throwable e){
logger.error(e.getMessage(),e);
}finally{
if(jedis!=null){
RedisManager.returnResource(jedis);
}
}
return task;
}
}4、获取队列实例的工具类
/**
* <pre>
* // 获得队列
* TaskQueue tq = TaskQueueManager.get(TaskQueueManager.SMS_QUEUE);
*
* // 添加任务到队列
* String task = "task id";
* tq.pushTask(task);
*
* // 从队列中取出任务执行
* String taskToDo = tq.popTask();
* </pre>
* @author liuhailong
*/
public class TaskQueueManager {
protected final static Logger logger = Logger.getLogger(TaskQueueManager.class);
private static Map<String, TaskQueueRedisImpl> queneMap = new ConcurrentHashMap<String, TaskQueueRedisImpl>();
/**
* 短信队列名。
*/
public static final String SMS_QUEUE = "SMS_QUEUE";
/**
* 规则队列名。
*/
public static final String RULE_QUEUE = "RULE_QUEUE";
private static void initQueneMap() {
logger.debug("初始化任务队列...");
queneMap.put(RULE_QUEUE, new TaskQueueRedisImpl(RULE_QUEUE));
logger.debug("建立队列："+RULE_QUEUE);
queneMap.put(SMS_QUEUE, new TaskQueueRedisImpl(SMS_QUEUE));
logger.debug("建立队列："+SMS_QUEUE);
}
static {
initQueneMap();
}
public static TaskQueue get(String name){
return getRedisTaskQueue(name);
}
public static TaskQueue getRedisTaskQueue(String name){
return queneMap.get(name);
}
}和具体的队列过于紧耦合，但简单好用。
先跑起来再说。
5、向队列中添加任务的代码
TaskQueue tq = TaskQueueManager.get(TaskQueueManager.SMS_QUEUE);
tq.pushTask(smsMessageId);6、从队列中取出任务执行的代码
public class SmsSendTask{
protected final static Logger logger = Logger.getLogger(SmsSendTask.class);
protected static SmsSendService smsSendService = new SmsSendServiceUnicomImpl();
/**
* 入口方法。
*/
public void execute() {
TaskQueue taskQueue = null;
String task = null;
try {
taskQueue = TaskQueueManager.get(TaskQueueManager.SMS_QUEUE);
// 非线程安全
Set<Serializable> executedTaskSet = new HashSet<Serializable>();
task = taskQueue.popTask();
while(task!=null){
// 判断是否把所有任务都执行一遍了，避免死循环
if(executedTaskSet.contains(task)){
taskQueue.pushTask(task);
break;
}
executeSingleTask(taskQueue,task);
task = taskQueue.popTask();
}
}catch(Throwable e){
logger.error(e.getMessage(),e);
e.printStackTrace();
}
}
/**
* 发送单条短信。
*
* 取出任务并执行，如果失败，放回任务列表。
*
* @param taskQueue
* @param task
*/
@SuppressWarnings({ "rawtypes", "unchecked" })
private void executeSingleTask(TaskQueue taskQueue, String task) {
try {
// do the job
String smsId = task;
Map<String,String> sms = smsSendService.getSmsList(smsId);
smsSendService.send(sms);
smsSendService.updateSmsStatus(task,SmsSendService.STATUS_SENT);
String opType = "2";
TaskQueueUtil.taskLog(taskQueue.getName(), opType, task);
} catch (Throwable e) {
if(task!=null){
taskQueue.pushTask(task);
smsSendService.updateSmsStatus(task,SmsSendService.STATUS_WAIT);
if(logger.isDebugEnabled()){
logger.error(String.format("任务%s执行失败：%s，重新放回队列", task, e.getMessage()));
}
}else {
e.printStackTrace();
}
}
}
}这部分代码是固定模式，而且不这样做存在重大缺陷，会有任务执行失败，被丢弃，这部分代码应该写到队列实现中。
有空再改。