[大数据]个人分析理解，项目到底需不需要用到MQ技术

个人分析理解，项目到底需不需要用到MQ技术

订单项 -> 订单规格-> 履行集 -> 履行流程-> OPU

上面是公司根据订单业务抽象成的订单框架流程结构。

这里的OPU是最小执行单元，它可以是同步也可以是异步。

由于该项目没有秒杀方面的需求，MQ就是用来跨系统的异步通信。

这里的异步用到过两种技术，一开始是多线程（线程池） + 定时任务（Quartz），后来改为了 RabbitMQ。

由于涉及保密条例，这里用伪代码来表示下不用MQ的情况下，是如何实现与外围系统的异步通信：

有几点要先注意下。

1. 使用线程池需要注意线程池的数量设置，套用公式 U * U *（1 + w/c）：CPU内核数 * 期待CPU利用率 * （1+ 单条线程等待时间/单条线程计算时间）
2. 定时任务的循环周期需要确定好，曾经项目中设置为1分钟，后来根据每天的订单数量和订单里异步类型的通信数量，改周期为5分钟
3. 消息发送失败的重试机制要想好，项目中一开始为了配套这块业务，主要是准备了两个重要的表：（1）订单号、存储报文、接口信息和重试次数等字段的quartz_info_request表 (2) 记录所有内调外或者外调内接口的日志信息表quartz_info_log ，里面含有订单号、接口信息等字段

/**
 * @Description:线程池配置
 * @author: 老街俗人
 */
@Configuration
public class ThreadPoolConfig {
@Bean
public ThreadPoolExecutor threadPoolExecutor{
    //核心线程数
    int corePoolSize = 50;
    //最大线程数
    int maximumPoolSize = 100;
    //超过 corePoolSize 线程数量的线程最大空闲时间
    long keepAliveTime =2;
    //以秒为时间单位
    TimeUnit unit = TimeUnit.SECONDS;
    //用于存放提交的等待执行任务
    BlokckingQueue<Runnable> workQueue = new ArrayBlockingQueue<Runnable>(40);
    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
    corePoolSize,maximumPoolSize,keepAliveTime,unit,workQueue,
        //默认策略，在拒绝任务时抛出RejectedExecutionException
        new ThreadPoolExecutor.AbortPolicy());
    );
    return threadPoolExecutor;
}
}
/**
 * @Description:异步请求
 * @author: 老街俗人
 */
public class SendService{
    @Autowired
    ThreadPoolExecutor threadPoolExecutor;
    
    public void asyncSend(){
        threadPoolExecutor.submit(()->{
            //1.调用对应的接口，发送请求
            //2.把相应信息记录到quartz_info_request表里，重试次数设置为0，是否成功设置为false
        })
    }
}
/**
 * @Description:定时任务里负责扫描quartz_info_request和quartz_info_log表，并比对哪些接口需要重试
 * @author: 老街俗人
 */
public class Sendjob{
    public void retry(){
        //扫描quartz_info_request和quartz_info_log表
        //根据订单号匹配
        //quartz_info_log没有匹配到到第三方响应接口返回来的信息，且已经超时，则重试接口；
        //如果明确响应接口已经报错，则把quartz_info_request对应的数据的重试次数置为最大次数，把当前订单信息录入异常表，用于对账和人工修单
        //......
        //实际上还有很多的步骤需要考虑，这里为了简洁明了，就写了几个主要步骤
    }
    
}

这样线程池 + 定时任务，也可以做到异步通信的结果，至于订单的对账和修复，全在于设计消息发送失败机制的多方面考虑。

明显不那么容易把控

接下来再看看用了RabbitMQ后，带来的影响。

首先是代码：

/**
 * @Description:MQ配置，这里使用的是直联交换机
 * @author: 老街俗人
 */
@Configuration
public class RabbitConfig {
//队列的名字是DirectQueue
@Bean
public Queue DirectQueue(){
    //durable 设置是否持久化，默认为false，true会存储在硬盘
    return new Queue("DirectQueue",true);
}
//交换机的名字是MyDirectExchange
@Bean
public DirectExchange MyDirectExchange(){
    //durable 设置是否持久化，默认为false，true会存储在硬盘
    return new DirectExchange("MyDirectExchange",true,false);
}
//绑定，将队列和交换机绑定
@Bean
public Binding bindingDirect(){
    //durable 设置是否持久化，默认为false，true会存储在硬盘
    return BindingBuilder.bind(DirectQueue()).to(MyDirectExchange()).with("MyDirectRouting");
}
/**
 * @Description:消息确认机制的配置
 * @author: 老街俗人
 */    
@Bean
public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){
    RabbitTemplate rabbitTemplate = new RabbitTemplate();
    rabbitTemplate.setConnectionFactory(connectionFactory);
    //开启Mandatory，触发回调函数
    rabbitTemplate.setMandatory(true);
    //这里针对的是生产者->broker(主机)消息传递失败的措施
    rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback(){
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        	System.out.println("ConfirmCallback:     "+"correlationData："+correlationData);
            System.out.println("ConfirmCallback:     "+"确认情况："+correlationData);
            System.out.println("ConfirmCallback:     "+"cause："+correlationData);
            //把correlationData记录到异常信息表，方便修单（确认模式下的回调函数，一般可能是网络问题或者磁盘满了，这些情况出现较少，配合日志监控系统，运维人员能即时修复）
        }
    });
    //这里针对的是交换机->队列消息传递失败的措施
    rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback(){
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey){
            	System.out.println("ReturnCallback:     "+"消息："+message);
				System.out.println("ReturnCallback:     "+"回应码："+replyCode);
                System.out.println("ReturnCallback:     "+"回应信息："+replyText);
                System.out.println("ReturnCallback:     "+"交换机："+exchange);
                System.out.println("ReturnCallback:     "+"路由键："+routingKey);
			//把message记录到异常信息表，方便修单（这种情况下的回调函数，一般可能是routingkey出错或者找不到队列，这些情况出现较少，生产环境基本不可能有）
        }
    });
    return rabbitTemplate;
}    
    
    
}

接下来是消费者手动确认的消息确认机制代码

@Configuration
public class RabbitConfig2 {
@Autowired
private CachingConnectionFactory connectionFactory;
@Autowired
private AckReceiver ackReceiver;//消息接收处理类
@Bean
public SimpleMessageListenerContainer simpleMessageListenerContainer(){
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
    container.setConcurrentConsumers(1);
    container.setMaxConcurrentConsumers(1);
    container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默认是自动确认，这里改为手动确认消息
    //设置一个队列
    container.setQueueNames("DirectQueue");
    container.setMessageListener(ackReceiver);
    return container;
} 
}

消费者代码

/**
 * @Description:消费者根据Message发送请求，如果出现异常失败，根据具体情况选择重发还是放弃
 * @author: 老街俗人
 */
@Component
public class AckReceiver implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //msg就是传递的消息
            String msg = message.toString();
            //从msg里取出对应信息，调用对应的接口，发送请求
            //多条信息可以配合for循环发送
            channel.basicAck(deliveryTag, true); //当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
        } catch (Exception e) {
            //可以更详细的划分Exception，根据不同异常
            //可以选择重新放回队列，如超时或者网络问题
            //失败不重发的要记录到异常订单表，方便对账和修单
			//channel.basicReject(deliveryTag, true);//当该参数为 true 时，该消息会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }
}

生产者代码

/**
 * @Description:异步请求
 * @author: 老街俗人
 */
public class SendService{
    @Autowired
    RabbitTemplate rabbitTemplate;
    
    public String asyncSend(){
    	Map<String,Object> map=new HashMap<>();
    	//加入多条键值对，存入相关信息，发送到broker
        //例如订单号、请求报文、接口名称等
        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);
        return "ok";
    }
}

总结一下：

使用MQ后，明显看到对于消息确认机制更加完善了，相当于用MQ代替了基于数据库的定时任务，效率更高了不说，对于开发人员，一旦出现消息异常或丢失，更容易排查。

只看分布式环境下的异步通信功能，MQ的引入还是很有必要的。

个人基于公司项目的理解，如有更好的看法，欢迎沟通~