二ActiveMQ JMS

Posted 2021-04-24 laughing Coding

一、Java Message Service

JMS即Java消息服务（Java Message Service）应用程序接口，是一个Java平台 中关于面向消息中间件 （MOM）的API ，用于在两个应用程序之间，或分布式系统 中发送消息，进行异步通信 。Java消息服务是一个与具体平台无关的API，绝大多数MOM提供商都对JMS提供支持。

二、JMS的四大元素

服务供应中间件provider（MQ服务器）

生产者producer

消费者consumer

消息message

三、消息message

消息头

JMSDestination消息发送目的地

textMessage.setJMSDestination(queue);

JMSDeliveryMode持久和非持久模式

textMessage.setJMSDeliveryMode(DeliveryMode.PERSISTENT);textMessage.setJMSDeliveryMode(DeliveryMode.NON_PERSISTENT);

JMSExpiration过期时间

textMessage.setJMSExpiration(6000);

JMSPriority消息优先级

0-9十个级别，0-4是普通消息，5-9是加急消息。不严格安装这个级别，但加急消息一定要先于普通消息。默认为4级。

textMessage.setJMSPriority(0);

JMSMessageID消息ID

唯一识别每个消息

textMessage.setJMSMessageID("xxxxxxx");

消息体

消息体有五种格式

session.createTextMessage();session.createMapMessage();session.createBytesMessage();session.createStreamMessage();session.createObjectMessage();

一般常用String和map

MapMessage mapMessage = session.createMapMessage();mapMessage.setString("K1","V1");

消息属性

实现识别、去重、重点标注等

textMessage.setStringProperty("PropertyName","PropertyValue");

四、持久化

持久化

// 非持久messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);// 持久化messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);

队列默认策略

对于queue队列 默认是持久化的

topic持久化

public class JmsTopicConsumer3 { public static final String ACTIVEMQ_URL = "tcp://49.232.61.212:61616"; public static final String TOPIC_NAME = "laughing-topic";
 public static void main(String[] args) throws JMSException, IOException {

 // 1、创建ActiveMQConnectionFactory ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL); // 2、创建connection 启动 Connection connection = activeMQConnectionFactory.createConnection(); connection.setClientID("laughing");
 // 3、创建session createSession(boolean transacted, int acknowledgeMode) // transacted 事务 ， acknowledgeMode 签收 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 4、创建目的地 队列 // Destination destination = session.createQueue(ACTIVEMQ_URL); Topic topic = session.createTopic(TOPIC_NAME); // 5、消费者
 // 订阅者 TopicSubscriber topicSubscriber = session.createDurableSubscriber(topic, "remark..."); connection.start();
 Message message = topicSubscriber.receive(); while (message != null) { TextMessage textMessage = (TextMessage) message; System.out.println("持久化topic:" + textMessage.getText()); message = topicSubscriber.receive(1000L);
 }
 session.close(); connection.close();
 }}

五、事务

保证高可用性

六、签收

签收有四种模式

手动签收需要客户端调用acknowledge手动签收

Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);