ActiveMQ之生产分组消息、消费分组消息

刺骨的言语ヽ痛彻心扉 2021-09-19 23:56 632阅读 0赞

今天突然遇到一个需求，就是要求我的ActiveMQ队列消息需要多个线程消费者去消费，但是每个消费者，虽然MQ支持多个消费者但是，每个消费者消费的数量可能不同，有的线程跑的快的，就肯定会多消费一些，但是现在的需求是要每个线程消费的数量一样，这就需要用到MQ分组了，

首先是在消息生产的时候就就对每个消息进行分组

生产者工具类

@Component
    public class ActiveMqUtils {
    	/**
    	 * 注入JMS
    	 */
    	@Autowired
    	private JmsTemplate jmsTemplate;
    
    	
    	public <T> String sendNorMolMessageById(Destination destination, String text,String GroupId) {
    		// 连接工厂
    		ConnectionFactory connectionFactory = jmsTemplate.getConnectionFactory();
    		Connection connection = null;
    		Session session = null;
    		MessageProducer producer = null;
    		try {
    			// 创建链接
    			connection = connectionFactory.createConnection();
    			connection.start();
    			// 创建session，开启事物
    			session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
    			// 创建生产者
    			producer = session.createProducer(destination);
    			// 设置持久化
    			producer.setDeliveryMode(DeliveryMode.PERSISTENT);
    			// 设置过期时间
    			//producer.setTimeToLive(time);
    			TextMessage message = session.createTextMessage(text);
    			message.setStringProperty("JMSGroupID",GroupId);
    			producer.send(message);
    			// 提交
    			session.commit();
    			return message.getJMSMessageID();
    		} catch (JMSException e) {
    			throw new RuntimeException(e);
    		} finally {
    			// 关闭连接
    			close(producer, session, connection, connectionFactory);
    		}
    	}

TextMessage message = session.createTextMessage(text);  
            **message.setStringProperty("JMSGroupID",GroupId);**

**然后**用工具生产，

/**
    	 * 产生点对点的消息队列
    	 */
    	@Override
    	public Result<String> createMq(String quenName,String text) {
    		Destination destination = new ActiveMQQueue(quenName);
    		String JMSMessageId = null;
    		for (int i = 0; i < 10; i++) {
    		/*Stu stu = stuMapper.selectByPrimaryKey(1);
    		String stuStr = new String();
    		if (stu!=null) {
    			stuStr = FastJsonUtils.getBeanToJson(stu);
    		}else {
    			stuStr = text;
    		}*/
    			JMSMessageId = utils.sendNorMolMessageById(destination, text, "GroupIdA");
    			JMSMessageId = utils.sendNorMolMessageById(destination, text, "GroupIdB");
    			//延时6秒后再次发送
    			//utils.sendDelayMessage(destination, text, 6*1000,6*1000,3);
    		}
    		return new Result<String>("消息已经产生："+JMSMessageId,true);
    	}

分别命名为GroupIdA，GroupIdB两个分组，

最后对消息进行监听

@Component
    public class ActiveMqListenConfig {
    
    	private Logger logger = LoggerFactory.getLogger(this.getClass());
    
    	 @JmsListener(destination="queueSend",selector="JMSGroupID='GroupIdA'")
    	public void recieveTaskMq(String message) {
    		logger.info("消费者1监听到的消息是：" + message + "，并且这条消息已经被消费了");
    	}
    
    	 @JmsListener(destination="queueSend" ,selector="JMSGroupID='GroupIdB'")
    	public void recieveTaskMq2(String message) {
    		logger.info("消费者2监听到的消息是：" + message + "，并且这条消息已经被消费了");
    	}
    
    }

这里值得注意一下的就是selector="JMSGroupID='GroupIdA'"，这是@JmsListener里面的，是可以根据条件来对消息进行监听的

在启动项目后调用接口

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTc4OTk4_size_16_color_FFFFFF_t_70][]

![20190610182130630.png][]![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTc4OTk4_size_16_color_FFFFFF_t_70 1][]

\-------------------------------------------------------------------------------------------------------------------------------------------------------------------------以上是我将两个消费者分到了不同的组所以它会将它所监听到的队列消息全部消费完，

当我让两个消费者全部监听GroupIdA时，两个消费者加起来肯定只会消息10跳消息，队列中肯定还会剩下10条都是GroupIdB的

@Component
    public class ActiveMqListenConfig {
    
    	private Logger logger = LoggerFactory.getLogger(this.getClass());
    
    	 @JmsListener(destination="queueSend",selector="JMSGroupID='GroupIdA'")
    	public void recieveTaskMq(String message) {
    		logger.info("消费者1监听到的消息是：" + message + "，并且这条消息已经被消费了");
    	}
    
    	 @JmsListener(destination="queueSend" ,selector="JMSGroupID='GroupIdA'")
    	public void recieveTaskMq2(String message) {
    		logger.info("消费者2监听到的消息是：" + message + "，并且这条消息已经被消费了");
    	}
    
    }

重启启动项目，跑起来

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTc4OTk4_size_16_color_FFFFFF_t_70 2][]

看下控制台，只有10条被消费了

再看下队列消息，![2019061018293658.png][]

也只被消费了10条，验证成功

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTc4OTk4_size_16_color_FFFFFF_t_70]: /images/20210708/340289500a8a48d7a262f5a6d0573e18.png
[20190610182130630.png]: https://img-blog.csdnimg.cn/20190610182130630.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTc4OTk4_size_16_color_FFFFFF_t_70 1]: /images/20210708/1443c32a924c4b028e7e659fb3a3b5de.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTc4OTk4_size_16_color_FFFFFF_t_70 2]: /images/20210708/eec74014ed1244cfa6e057195cda3304.png
[2019061018293658.png]: /images/20210708/fff7ce6ddfee41ddb6bd3dbef87ddc45.png