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

1.资源管理

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

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

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

7.日志追踪

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

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

如何减少连接数？

在发送大批量消息的情况下，创建和释放连接依然有不小的开销。我们可以跟接收方约定批量消息的格式，比如支持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），一次发送的消息数需要合理地控制。