基于Redis队列实现《二》——点对点模式

前言

       通过队列实功能之间的解耦，异步，是一个非常通用的功能。一般采用 请求——>异步处理——>异步通知的方式实现，比如支付宝支付，提交支付请求之后，直接跳转一个中间界面，过几秒钟后，才会提示支付成功。

       这里实现的场景是为了解耦，在提交商品购买请求之后，需要做很多事情，比如保存请购单数据，生成订单数据，减扣库存，提交OA审批，还有消息通知等等，如果采用传统的同步请求的方式，这个接口可能会需要

等待比较长的时间才会返回结果，会给用户带来是不是很好的体验。这里在生成请购单和订单数据后，以消息的形式推送redis队列，异步减扣库存，提交OA审批等操作。

队列基础

    这里可以参考前一篇 《基于Redis队列实现《一》——SUB/PUB模式》

示例

1. 实现说明：这里只是借签RedisTemplate实现redis队列的pub/sub模式的设计，基于spring 容器 linstener做的一个实现。这里实现的是点对点模式。

2. 知识准备：

     2.1  这里基于spring boot 项目，通过ResdisTemplate客户端连接redis;

     2.2  该方案依赖于redis数据结构List，通过LPUSH(左侧插入)和BLPOP（阻塞出队）实现的点对点阻塞队列。

     2.3  spring容器自定义listener机制

3.实现流程：

 

4. 代码示例:

 1. RedisQuenueConfig配置类

/**
 * Redis配置类
 * @version 1.0
 * @author liuyq
 * @date 2021-06-03 15:17:55
 */
@Configuration
public class RedisQuenueConfig {

    /**
     * 实例化listener
     * @return
     */
    @Bean
    public RecEventListener getEventListener(){
        return new RecEventListener();
    }
}

2.RecEventListener 事件监听器，当redis队列收到消息时，将消息出队

/**
 * RecEvent事件监听器
 * @version 1.0
 * @author liuyq
 * @date 2021-06-03 14:26:22
 */
@Component
public class RecEventListener implements ApplicationListener<RecEvent> {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private MessageService messageService;

    @Override
    public void onApplicationEvent(RecEvent recEvent) {
        ListOperations<String, String> stringStringListOperations = stringRedisTemplate.opsForList();
        String msg = stringStringListOperations.rightPop(recEvent.getQueue());
        if(msg == null){
            return;
        }
        //处理指定channel的消息
        if(Constant.third_order_message_queue_prefix.equals(recEvent.getQueue())){
            messageService.handleMessage(msg); 
        }

   }
 }

3. RecEvent 接收消息事件

/**
 * 接收消息事件
 * @version 1.0
 * @author liuyq
 * @date 2021-06-03 14:21:11
 */
public class RecEvent extends ApplicationEvent {

    private String queue;

    private String msg;

    public RecEvent(Object source) {
        super(source);
    }

    public RecEvent(Object source, String queue, String msg) {
        super(source);
        this.queue = queue;
        this.msg = msg;
    }

    public String getQueue() {
        return queue;
    }

    public void setQueue(String queue) {
        this.queue = queue;
    }

    public String getMsg() {
        return msg;
    }

    public void setMsg(String msg) {
        this.msg = msg;
    }
}

　　

4. PbulishMessageServiceImp 客户端发布消息

/**
 * 发布消息
 * @version 1.0
 * @author liuyq
 * @date 2021-04-29 17:22:02
 */
@Service
public class PbulishMessageServiceImpl implements PbulishMessageService {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private WebApplicationContext webapplicationcontext;


    /**
     * 保存消息到队列，基于点对点模式
     * @param queue 队列列
     * @param content 内容
     */
    public void saveMessageToQueueWithSingle(String queue, String content){
        ListOperations<String, String> listOperations = stringRedisTemplate.opsForList();
        Long result = listOperations.leftPush(queue, content);
        if(result > 0){
            RecEvent recEvent = new RecEvent("object", Constant.third_order_message_queue_prefix, "####监听第三方订单创建");
            //推送消息后，发布时间RecEvent，告知监听器处理消息
            webapplicationcontext.publishEvent(recEvent);
        }
    }

}

5. 调用示例

@Service
public class Demo{

    @Autowired
    private MessageService messageService;

    /**
    *   推送消息示例
    *  @param channel  消息队列名
    *  @param msg 推送消息
    */
    public void publishMessage(String channel, String msg){
       //推送到redis队列，之前由于1对多问题导致停滞     
       messageService.saveMessageToQueueWithSingle(Constant.third_order_message_queue_prefix, msg);
    }
}