ActiveMQ持久订阅设置

我不是女神ヾ 2024-04-06 14:04 60阅读 0赞

## ActiveMQ持久订阅设置 ##

在JMS中,Topic实现publish和subscribe语义。一条消息被publish时,它将发到所有感兴趣的订阅者,所以零到多个subscriber将接收到消息的一个拷贝。但是在消息代理接收到消息时,只有激活订阅的subscriber能够获得消息的一个拷贝。  
JMS Queue执行load balancer语义。一条消息仅能被一个consumer收到。如果在message发送的时候没有可用的consumer,那么它将被保存一直到能处理该message的consumer可用。如果一个consumer收到一条message后却不响应它,那么这条消息将被转到另一个consumer那儿。一个Queue可以有很多consumer,并且在多个可用的consumer中负载均衡

#### queue与topic的技术特点对比 ####

Topic Queue  
概要 Publish Subscribe messaging 发布订阅消息 Point-to-Point 点对点  
有无状态 topic数据默认不落地，是无状态的。 Queue数据默认会在mq服务器上以文件形式保存，比如Active MQ一般保存在$AMQ\_HOME\\data\\kr-store\\data下面。也可以配置成DB存储。  
完整性保障 并不保证publisher发布的每条数据，Subscriber都能接受到。  
Queue保证每条数据都能被receiver接收。

消息是否会丢失 一般来说publisher发布消息到某一个topic时，只有正在监听该topic地址的sub能够接收到消息；如果没有sub在监听，该topic就丢失了。 Sender发送消息到目标Queue，receiver可以异步接收这个Queue上的消息。Queue上的消息如果暂时没有receiver来取，也不会丢失。

消息发布接收策略 一对多的消息发布接收策略，监听同一个topic地址的多个sub都能收到publisher发送的消息。Sub接收完通知mq服务器 一对一的消息发布接收策略，一个sender发送的消息，只能有一个receiver接收。receiver接收完后，通知mq服务器已接收，mq服务器对queue里的消息采取删除或其他操作。  
Topic和queue的最大区别在于topic是以广播的形式，通知所有在线监听的客户端有新的消息，没有监听的客户端将收不到消息；而queue则是以点对点的形式通知多个处于监听状态的客户端中的一个。

ActiveMQ持久订阅设置

通过为发布者设置 PERSISTENT传送模式，为订阅者时使用持久订阅，这样可以保证 Pub/Sub 程序接收所有发布的消息。

消息订阅分为非持久订阅(non-durable subscription)和持久订阅(durable subscription)，非持久订阅只有当客户端处于激活状态，也就是和 ActiveMQ 保持连接状态才能收到发送到某个主题的消息，而当客户端处于离线状态，这个时间段发到主题的消息将会丢失，永远不会收到。持久订阅时，客户端向ActiveMQ 注册一个识别自己身份的 ID，当这个客户端处于离线时，ActiveMQ会为这个 ID 保存所有发送到主题的消息，当客户端再次连接到ActiveMQ 时， 会根据自己的 ID 得到所有当自己处于离线时发送到主题的消息。持久订阅会增加开销，同一时间在持久订阅中只有一个激活的用户。

#### 建立持久订阅的步骤： ####

1． 为连接设置一个客户 ID；  
2． 为订阅的主题指定一个订阅名称；  
上述组合必须唯一。

#### 代码实现持久化订阅 ####

（1）使用queue，即队列时，每个消息只有一个消费者，所以，持久化很简单，只要保存到数据库即可

。然后，随便一个消费者取走处理即可。某个消费者关掉一阵子，也无所谓。

（2）使用topic，即订阅时，每个消息可以有多个消费者，就麻烦一些。

##### 首先，假设消费者都是普通的消费者 #####

<1>activemq启动后，发布消息1，可惜，现在没有消费者启动着，也就是没有消费者进行了订阅。那么

，这个消息就被抛弃了。

<2>消费者1启动了，连接了activemq，进行了订阅，在等待消息~~

activemq发布消息2，OK，消费者1收到，并进行处理。消息抛弃。

<3>消费者2也启动了，连接了activemq，进行了订阅，在等待消息~~

activemq发布消息3，OK，消费者1，消费者2都收到，并进行处理。消息抛弃。

<4>消费者1关掉了。

activemq发布消息4，OK，消费者2收到，并进行处理。消息抛弃。

<5>消费者1又启动了。

##### activemq发布消息5，OK，消费者1，消费者2都收到，并进行处理。消息抛弃。 #####

总结一下：  
activemq只是向当前启动的消费者发送消息。  
关掉的消费者，会错过很多消息，并无法再次接收这些消息。

如果发送的消息是重要的用户同步数据，错过了，用户数据就不同步了。

那么，如何让消费者重新启动时，接收到错过的消息呢？

### 答案是持久订阅。 ###

普通的订阅，不区分消费者，场地里有几个人头，就扔几个馒头。  
持久订阅，就要记录消费者的名字了。  
张三说，我是张三，有馒头给我留着，我回来拿。  
李四说，我是李四，有馒头给我留着，我回来拿。  
activemq就记下张三，李四两个名字。

那么，分馒头时，还是一个人头给一个馒头。  
分完了，一看张三没说话，说明他不在，给他留一个。  
李四说话了，那就不用留了。

张三回来了，找activemq，一看，这不张三吧，快把他的馒头拿来。  
可能是一个馒头，也可能是100个馒头，就看张三离开这阵子，分了多少次馒头了。

##### activemq区分消费者，是通过clientID和订户名称来区分的。 #####

// 创建connection  
connection = connectionFactory.createConnection();  
connection.setClientID(“bbb”); //持久订阅需要设置这个。  
connection.start();

// 创建session  
Session session = connection.createSession(false,Session.AUTO\_ACKNOWLEDGE);

// 创建destination  
Topic topic = session.createTopic(“userSyncTopic”); //Topic名称

//MessageConsumer consumer = session.createConsumer(topic); //普通订阅  
MessageConsumer consumer = session.createDurableSubscriber(topic,“bbb”); //持久订阅

还有一点，消息的生产者，发送消息时用使用持久模式  
MessageProducer producer = …;  
producer.setDeliveryMode(DeliveryMode.PERSISTENT);  
不设置，默认就是持久的

（5）使用相同的“clientID”，则认为是同一个消费者。两个程序使用相同的“clientID”，则同时只能有一个连接到activemq，第二个连接的会报错。

（6）activemq的设置在conf/activemq.xml中，默认消息是保存在data/kahadb中，重启activemq消息不会丢。

可以访问http://localhost:8161/admin/index.jsp：查看当前的队列、Topic和持久订户的信息、发送消息等等，很方便。

可以复制activemq-jdbc.xml中的内容过来，修改一下，就可以把消息保存在其它数据库中了。

系统启动卡顿的解决办法

<bean id="amqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    	<property name="brokerURL" value="tcp://localhost:61616"></property>
    	<property name="userName" value="admin"></property>
    	<property name="password" value="admin"></property>
    </bean>