activemq实现队列的独有消费

柔情只为你懂 2024-02-19 18:23 35阅读 0赞

在我们实际的开发中可能存在这么一种情况，应用程序要向一个**队列**名为queue的队列中发送3条消息，需要保证这3条消息按顺序消费。必须是第一条消费完，在消费第二条然后是第三条。而我们的程序中可能有时候存在多个consumer对这个队列进行消费，那么可能出现消息时按1,2,3进行消费的，但第二条可能比较耗时，就会导致第2条消息没有消费完，第三条消息就已经消费完了，这个时候可能就会出现问题。

**解决方案**：1、使用独有消费者就可以解决这个问题。（在创建队列的时候，向队列后面增加参数 **?consumer.exclusive=true**）

2、使用消息分组也可以进行解决（比上面一种方案好）

3、本博文使用的是独有消费者进行解决

**场景模拟**：向队列中发送3条消费，且在消费第二条消费时，线程沉睡5秒中，程序中启动2个监听对消息进行消费，看消息的消费结果

队列是属于点对点的模式，且队列中的一条消费只可能被一个消费者进行消费

1、pom文件的配置

<dependency>
    	<groupId>org.apache.activemq</groupId>
    	<artifactId>activemq-all</artifactId>
    	<version>5.11.1</version>
    </dependency>
    <dependency>
    	<groupId>org.apache.activemq</groupId>
    	<artifactId>activemq-pool</artifactId>
    	<version>5.12.1</version>
    </dependency>
    <dependency>
    	<groupId>org.springframework</groupId>
    	<artifactId>spring-jms</artifactId>
    	<version>4.3.8.RELEASE</version>
    </dependency>
    <dependency>
    	<groupId>org.springframework</groupId>
    	<artifactId>spring-aop</artifactId>
    	<version>4.3.8.RELEASE</version>
    </dependency>
    <dependency>
    	<groupId>org.springframework</groupId>
    	<artifactId>spring-beans</artifactId>
    	<version>4.3.8.RELEASE</version>
    </dependency>
    <dependency>
    	<groupId>org.springframework</groupId>
    	<artifactId>spring-context</artifactId>
    	<version>4.3.8.RELEASE</version>
    </dependency>
    <dependency>
    	<groupId>org.springframework</groupId>
    	<artifactId>spring-context-support</artifactId>
    	<version>4.3.8.RELEASE</version>
    </dependency>
    <dependency>
    	<groupId>org.springframework</groupId>
    	<artifactId>spring-core</artifactId>
    	<version>4.3.8.RELEASE</version>
    </dependency>
    <dependency>
    	<groupId>org.springframework</groupId>
    	<artifactId>spring-messaging</artifactId>
    	<version>4.3.8.RELEASE</version>
    </dependency>
    <dependency>
    	<groupId>org.springframework</groupId>
    	<artifactId>spring-oxm</artifactId>
    	<version>4.3.8.RELEASE</version>
    </dependency>
    <dependency>
    	<groupId>org.springframework</groupId>
    	<artifactId>spring-tx</artifactId>
    	<version>4.3.8.RELEASE</version>
    </dependency>
    <dependency>
    	<groupId>commons-pool</groupId>
    	<artifactId>commons-pool</artifactId>
    	<version>1.6</version>
    </dependency>

2、消息监听器一 (消费者一)

public class ExclusiveConsumerListener implements SessionAwareMessageListener<TextMessage> {
    	@Override
    	public void onMessage(TextMessage message, Session session) throws JMSException {
    		String msg = message.getText();
    		if (msg.startsWith("2")) {
    			try {
    				System.out.println(Thread.currentThread().getName() + " 开始沉睡5秒.");
    				Thread.sleep(5000);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    		System.out.println(Thread.currentThread().getName() + " : " + this.getClass().getName() + " :开始消费消息:" + msg);
    	}
    }

3、消息监听器二（消费者二）

public class ExclusiveConsumerListener2 implements SessionAwareMessageListener<TextMessage> {
    	@Override
    	public void onMessage(TextMessage message, Session session) throws JMSException {
    		String msg = message.getText();
    		if (msg.startsWith("2")) {
    			try {
    				System.out.println(Thread.currentThread().getName() + " 开始沉睡5秒.");
    				Thread.sleep(5000);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    		System.out.println(Thread.currentThread().getName() + " : " + this.getClass().getName() + " :开始消费消息:" + msg);
    	}
    }

4、配置文件（**没有实现消息的独有消费**）

5、测试

public static void main(String[] args) {
    		ApplicationContext ctx = new ClassPathXmlApplicationContext("classpath:application-exclusive-consumer.xml");
    		JmsTemplate jmsTemplate = ctx.getBean(JmsTemplate.class);
    		String msg1 = "1-这是第一条消息";
    		String msg2 = "2-这是第二条消息";
    		String msg3 = "3-这是第三条消息";
    		jmsTemplate.convertAndSend(msg1);
    		jmsTemplate.convertAndSend(msg2);
    		jmsTemplate.convertAndSend(msg3);
    	}

6、测试结果（可以看到消息是乱的，并没有按照1消费完后消费2,2消费完后消费3）

![a2032095-80f3-3a95-a911-6cc135ca36ec.png][]

7、修改配置文件（在队列的字符串后面加上**?consumer.exclusive=true**）**实现消费的独有消费**

![6b6f44d8-fa51-33d4-83ba-b7bead6dffe7.png][]  
8、重新测试（**结果为按照顺序消费的**）

![926dd754-31a3-3f4e-9b43-5c2759559bab.png][]

[a2032095-80f3-3a95-a911-6cc135ca36ec.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/8cba61da7d2e4589a57831a11c194e81.png
[6b6f44d8-fa51-33d4-83ba-b7bead6dffe7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/f32c7d706b45496b97188f601167bd35.png
[926dd754-31a3-3f4e-9b43-5c2759559bab.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/deb2ad1a6b69459aba197619f1385bc8.png