基于Redis实现分布式消息队列(3)

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);1
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();
            }
        }
    }

}

这部分代码是固定模式,而且不这样做存在重大缺陷,会有任务执行失败,被丢弃,这部分代码应该写到队列实现中。
有空再改。

Redis 的详细介绍请点这里
Redis 的下载地址请点这里

内容版权声明:除非注明,否则皆为本站原创文章。

转载注明出处:https://www.heiqu.com/ec06f2f218266cb8351886e8b9dcbdbf.html