消息队列activeMQ使用

本文共 2543 字，大约阅读时间需要 8 分钟。

消息队列中间件是分布式系统中重要的组件，主要解决应用耦合，异步消息，流量削锋等问题

实现高性能，高可用，可伸缩和最终一致性架构

现在比较常见的MQ产品主要是ActiveMQ、RabbitMQ、ZeroMQ、Kafka、MetaMQ、RocketMQ，当然还有很多其他的产品。

本人使用的为activeMQ

从官网下载 解压  运行  \apache-activemq-5.15.4\bin\win64\activemq.bat

(本人使用的是5.15.4版本的，最新版本到5.15.5)、

访问activemq管理页面   http://localhost/8161/admin    出现activeMq管理页面即启动成功（默认用户名密码都为：admin）

activeMQ有两种消息模型P2P（Point to Point）,Publish/Subscribe(Pub/Sub)。

P2P

P2P模式包含三个角色：消息队列（Queue），发送者(Sender)，接收者(Receiver)。每个消息都被发送到一个特定的队列，接收者从队列中获取消息。队列保留着消息，直到他们被消费或超时。

P2P的特点

每个消息只有一个消费者（Consumer）(即一旦被消费，消息就不再在消息队列中)

发送者和接收者之间在时间上没有依赖性，也就是说当发送者发送了消息之后，不管接收者有没有正在运行，它不会影响到消息被发送到队列

接收者在成功接收消息之后需向队列应答成功 

如果希望发送的每个消息都会被成功处理的话，那么需要P2P模式。

Pub/sub模式

包含三个角色主题（Topic），发布者（Publisher），订阅者（Subscriber） 多个发布者将消息发送到Topic,系统将这些消息传递给多个订阅者。

Pub/Sub的特点

每个消息可以有多个消费者

发布者和订阅者之间有时间上的依赖性。针对某个主题（Topic）的订阅者，它必须创建一个订阅者之后，才能消费发布者的消息

为了消费消息，订阅者必须保持运行的状态

为了缓和这样严格的时间相关性，JMS允许订阅者创建一个可持久化的订阅。这样，即使订阅者没有被激活（运行），它也能接收到发布者的消息。

如果希望发送的消息可以不被做任何处理、或者只被一个消息者处理、或者可以被多个消费者处理的话，那么可以采用Pub/Sub模型。

其中发布订阅模式中包含：消息持久化和消息非持久化

持久传输和非持久传输最大的区别是：采用持久传输时，传输的消息会保存到磁盘中(messages are persisted to disk/database)，即“存储转发”方式。先把消息存储到磁盘中，然后再将消息“转发”给订阅者。当Borker宕机 恢复后，消息还在。

采用非持久传输时，发送的消息不会存储到磁盘中。Borker宕机重启后，消息丢失。

可以通过MessageProducer 类的 setDeliveryMode方法设置传输模式：

messageProducer .setDeliveryMode(DeliveryMode.PERSISTENT);

发送端代码

// 创建connection connection = connectionFactory.createConnection(); connection.start();

// 创建session Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);

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

MessageProducer producer = session.createProducer(topic); //创建生产者

  MapMessage mapMessage = session.createMapMessage();

//存放具体消息内容   mapMessage.setStringProperty("update","dimension");

producer.send(mapMessage);

接收端代码：

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);

4、接收消息

（1）同步

订阅者或接收者通过receive方法来接收消息，receive方法在接收到消息之前（或超时之前）将一直阻塞；

（2）异步

订阅者或接收者可以注册为一个消息监听器。当消息到达之后，系统自动调用监听器的onMessage方法。

转载地址：https://blog.csdn.net/qq_38130280/article/details/81872229 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！