【RabbitMQ】高级使用：实际开发中的几点建议

ゝ一世哀愁。 2022-12-22 04:57 182阅读 0赞

## 1.资源管理 ##

*  在实际生产中，交换机和队列实际上是作为资源（跑起来才有的），由运维管理员提前创建的。是要申请的
 *  但写代码时，还要声明（new）相应的Exchange、Queue、绑定
    
     *  为了保证所到Exchange，Queue一定存在
     *  若已经被创建好的Exchange，Queue再在代码new时，直接返回true，不报错
 *  Exchange，Queue 一般是在消费者的代码中定义
    
     *  谁使用是管理原则，所以一般在消费者创建
     *  同时还因为消费者启动就会开始监听队列，并且长时间在线

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center][]

## 2.命名规范 ##

1.  命名体现元数据的类型
    
     *  虚拟机命名： XXX\_VHOST
     *  交换机命名：XXX\_EXCHANGE
     *  队列命名：\_QUEUE
2.  命名体现数据来源和去向
    
    示例：销售系统发往产品系统的交换机：SALE\_TO\_PRODUCT\_EXCHANGE。做到见名知义，不用去查文档（当然注释是必不可少的）

## 3.配置文件+声明 ##

*  元数据的命名集中放在 properties 文件中，不要用硬编码。
 *  如果有多个系统，可以配置多个xxx\_mq.properties。

# mq.properties
    com.my.directexchange=MY_DIRECT_EXCHANGE
    com.my.topicexchange=MY_TOPIC_EXCHANGE
    com.my.fanoutexchange=MY_FANOUT_EXCHANGE
    com.my.firstqueue=MY_FIRST_QUEUE
    com.my.secondqueue=MY_SECOND_QUEUE
    com.my.thirdqueue=MY_THIRD_QUEUE
    com.my.fourthqueue=MY_FOURTH_QUEUE

@Configuration
    @PropertySource("classpath:mq.properties") // 读取properties文件
    public class RabbitConfig {
        
        @Value("${com.my.firstqueue}")  // 获取队列名
        private String firstQueue;
        
        @Value("${com.my.directexchange}") // 获取交换机名
        private String directExchange;
        
        @Bean("myFirstQueue")  // 声明队列
        public Queue getFirstQueue(){
        
            return new Queue(firstQueue);
        }
        
        @Bean("myDirectExchange") // 声明交换机
        public DirectExchange getDirectExchange(){
        
            return new DirectExchange(directExchange);
        }
        
        @Bean
        public Binding bindFirst(@Qualifier("myFirstQueue") Queue queue, // 声明绑定
                                 @Qualifier("myDirectExchange") DirectExchange exchange){
        
            return BindingBuilder.bind(queue).to(exchange).with("my.best");
        }
    
        
    }

## 4.再次封装 ##

*  在项目中可以对Template做进一步封装，简化消息的发送
    
    例如：如果交换机、路由键是固定的，封装之后就只需要一个参数：消息内容。
 *  另外，如果想要平滑地迁移不同的MQ（如果有这种需求的话），也可以再做一层简单的封装

sendMsg(){
        
    	JmsTemplate.send(destination,msg); 
    }
    // 这时，如果要把ActiveMQ替换为RabbitMQ，还是使用send方法，只用更改调用对象
    sendMsg(){
        
    	RabbitTemplate.send(exchange,routingKey,msg);
    }

## 5.信息落库+定时任务 ##

将需要发送的消息保存在数据库中，可以实现消息的可追溯和重复控制，需要配合定时任务来实现。

1.  将需要发送的消息登记在消息表中
2.  定时任务一分钟或半分钟扫描一次，将未发送的消息发送到MQ服务器，并且修改状态为已发送
3.  如果需要重发消息，将指定消息的状态修改为未发送即可

副作用：降低效率，浪费存储空间

## 6.生产环境运维监控 ##

虽然RabbitMQ提供了一个简单的管理界面，但是如果对于系统性能、高可用和其他参数有一些定制化的监控需求的话，我们就需要通过其他方式来实现监控了。

主要关注：磁盘、内存、连接数

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 2]

## 7.日志追踪 ##

RabbitMQ可以通过[Firehose][]功能来记录消息流入流出的情况，用于调试，排错。

*  它是通过创建一个TOPIC类型的交换机（amq.rabbitmq.trace），把生产者发送给Broker的消息或者Broker发送给消费者的消息发到这个默认的交换机上面来实现的。
 *  另外RabbitMQ也提供了一个Firehose的GUI版本，就是Tracing插件。启用Tracing插件后管理界面右侧选项卡会多一个Tracing，可以添加相应的策略。
 *  RabbitMQ还提供了其他的[插件][Link 1]来增强功能。

## 如何减少连接数？ ##

在发送大批量消息的情况下，创建和释放连接依然有不小的开销。我们可以跟接收方约定**批量消息的格式**，比如支持JSON数组的格式，通过**合并消息内容**，可以减少生产者/消费者与Broker的连接。

// Producer将MQ消息分成10个一条记录，并批量记录数据库
    private void partAndInsertMq(List<StoreProdToAlsBean> list, String dType) {
        
    	// 通过getAverRange方法，将没每10条数据合并成一组，表示1条消息
    	// 所以，最后会有n个条消息（每条消息里面10条数据），再批量发送
    	List<List<StoreProdToAlsBean>> partLists = getAverRange(list,10);
    	List<MQTask> mqList = new ArrayList<MQTask>();
    	for( List<SimsStoreProdToAlsBean> partList: partLists ){
        
    		
    		MQTask mqMsg = new MQTask();
    		mqMsg.setRoutingKey(Constant.MQRoutingKeys.MQ_CXF.getRoutingKey());
    		
    		JSONObject pdDelDson = new JS0N0bject();
    		pdDelDson.put("data", JSONArray.fromObject(partList));
    		pdDelDson.put("title", "storeProd");
    		pdDelDson.put("dType", dType);
    		mqMsg.setJsonDatastr(pdDelJson.toString〇);
    		
    		Long regDay = Long.porselong(DatesUtil.dateToString(new Date(), DatesUtil.SIMPLE_TIME));
    		mqMsg.setRegDay(regDay);
    		mqMsg.setLastUpTms(regDay);
    		mqMsg.setMqStatus("0");
    		
    		mqList.add(mqMsg);
    	}
    	NewBatchImportDao.batachInsertMQTask(mqlist);
    }
    
    /**
    * 把一个数组没份N个划分，返回一个最大的数组
    * @Param list 数组
    * @Param average 每份包含多少个
    * @Return 新的数组
    */
    public <T> List<List<T>> getAverRange( List<T> list, int average ){
         
    	List<List<T>> newArrays = new ArrayList<List<T>>(); 
    	int start = 0; 
    	int end = 0;
    	for (int i = 0; i <= list.size() / average; i++) {
        
    		List<T> partList = new ArrayList<T>(); 
    		end = start + average; 
    		if (start >= list.size〇) {
         
    			break;
    		}
    		if (end >= list.size()) {
         
    			end = list.size();
    		}
    		partList = list.subList(start, end); 
    		newArrays.add(partList); 
    		start = end;
    	}
    	return newArrays;
    }

等发送到Broker后，每条消息的格式如下，在数据部分是一个数组：

msgContent = {
         
        "title":"storeProd", 
        "dType":“add”, 
        "data":[ 
            {
        "merchName":"黄金手机店","address":"黄金路 999 号"}, 
            {
        "merchName":"银星手机店","address":"银星路 168 号"}, 
            {
        "merchName":"青山手机店","address":"青山路 73 号"} 
        ] 
    }

> 建议单条消息不要超过4M（4096KB），一次发送的消息数需要合理地控制。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/7d7f695fda8c47fb961fb45e39951e63.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/617803cc05df4e17993188f198509aa0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/2b5c4cef2b174c259079a624bdf31b1b.png
[Firehose]: https://www.rabbitmq.com/firehose.html
[Link 1]: https://www.rabbitmq.com/plugins.html