JMS概述

Posted 2020-12-04 helloz

介绍

Java Message Service是一组支持网络上主机间正式的消息通信的API， 它为Java应用程序提供了通用的消息协议以及消息服务。使用Jms可以创建，发送，读取消息。
JMS主要的作用是提供异步通信，实现应用程序组件间的解耦，同时提高的系统的可靠性， 稳定性。

消息通信模型

P2P模型(Poing-to-Poing)

在p2p模型中， 消息通信的实现依赖于消息队列，发送者，接收者。每条消息被发送到特定的队列之中， 队列会一直保存消息，直到消息被接收或者过期， p2p的特点如下：

• 对于每一条消息只有一个消费者
• 生产者，消费者对一条消息的处理没有时间上的同步要求
• 不管消费者是否在运行， 生产者发送消息之后消费者在消息未过期的情况下可随时接收消息
• 消费者接收完消息之后发送确认消息，确认消息已被接收。

发布/订阅模型

在发布/订阅模型中，一条消息通过一个主题(Topic)发送到所有的客户端，即订阅者。 发布者和订阅者是异步的， 并且可以自由向主题发布消息或订阅主题中的消息。 主题负责保存和传递消息。
发布/订阅模型的特点如下：

• 对于一条消息可以有多个订阅者，即多个消费者
• 消费者只有在订阅主题之后才能获得发布到主题上的消息， 并且订阅者必须一直保持活跃以接收消息

消息处理方式

在JMS中，消息处理可以是同步的或者异步的。

同步消息处理

在同步消息处理中， 消费者或者订阅者通过调用receive()方法接收消息， receive()方法会一直阻塞， 指导收到消息或者超过指定的时间。

异步消息处理

在异步消息处理中，订阅者和消费者可以注册一个消息监听器， 类似于事件监听器，当消息到达之后， JMS Provide会通过调用消息监听器的onMessage()方法传递消息。

JMS主要对象介绍

• Connection Factories and Destination
• Connections
• Sessions
• Message Producers
• Message Consumers
• Message listeners

JMS Administered Objects

Connection Factories
可以通过connection factory 创建一个连接， 即在JMS提供者和客户端之间创建连接。 JMS客户端可以在JNDI中查找连接， 这个连接即为服务端和客户端之间的连接。通过这个连接， 客户端可以通过队列或主题的方式发送接收消息。

QueueConnectionFactory queueConnFactory = (QueueConnectionFactory) initialCtx.lookup ("primaryQCF");
Queue purchaseQueue = (Queue) initialCtx.lookup ("Purchase_Queue");
Queue returnQueue = (Queue) initialCtx.lookup ("Return_Queue");

Destination
客户端可以通过指定destination来决定发送到哪条队列之上， 从而决定谁可以接收它。 JMS提供两种类型的destination， 队列(Queue)和主题(Topic)。

QueueSession ses = con.createQueueSession (false, Session.AUTO_ACKNOWLEDGE);  //get the Queue object 
Queue t = (Queue) ctx.lookup ("myQueue");  //create QueueReceiver 
QueueReceiver receiver = ses.createReceiver(t);

TopicSession ses = con.createTopicSession (false, Session.AUTO_ACKNOWLEDGE); // get the Topic object 
Topic t = (Topic) ctx.lookup ("myTopic");  //create TopicSubscriber 
TopicSubscriber receiver = ses.createSubscriber(t);

Connection

连接对象封装了服务端和客户端的连接， 它实现了Connection接口， 可通过ConnectionFactory对象创建连接，使用完之后可以关闭它。

Connection connection = connectionFactory.createConnection();
connection.close();

JMS Session

会话表示一个单线程的上下文用来发送或接收消息，通过会话可以创建生产者和消费者。 可以通过连接对象来创建会话

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

JMS Message Producer

生产者即由会话创建的用来发送消息的对象， 它实现了MessageProducer接口。

MessageProducer producer = session.createProducer(dest);
MessageProducer producer = session.createProducer(queue);
MessageProducer producer = session.createProducer(topic);
producer.send(message);

JMS Message Consumer

消费者即由会话创建的用来接收消息的对象，它实现了MessageConsumer接口， 可以使用它来接收destionation, queue或topic中的消息( destination是一个通用表示消息中介的接口， queue和topic实现了destination接口)。

MessageConsumer consumer = session.createConsumer(dest);
MessageConsumer consumer = session.createConsumer(queue);
MessageConsumer consumer = session.createConsumer(topic);

JMS Message Listeners

消息监听器是一个异步处理消息的对象， 它实现了MessageListener接口， 接口中有一个onMessage()方法用来定义处理动作， 通过在消费者上调用setMessageListener()方法来设置消息监听器。

Listener myListener = new Listener();
consumer.setMessageListener(myListener);

JMS消息组成

1. Message Header
   包含一系列属性被服务端和客户端用来发布及识别消息， 这些属性包括：
   – JMSDestination
   – JMSDeliveryMode
   – JMSMessageID
   – JMSTimestamp
   – JMSCorrelationID
   – JMSReplyTo
   – JMSRedelivered
   – JMSType
   – JMSExpiration
   – JMSPriority

2. Message Properties
   应用程序可以创建，设置消息属性，这些属性是一系列键值对。 通过读取消息属性可以过滤消息， JMS也提供了一些预定义的服务端可以识别的消息属性， 这些属性是可选的。

3. Message Body
   JMS预定义五种不同的消息体类型来发送或接收消息， 包括：

• Text message : javax.jms.TextMessage对象， 代表文本消息
• Object message: javax.jms.ObjectMessage对象, 代表java对象
• Bytes message: javax.jms.BytesMessage对象， 代表二进制消息
• Stream message: javax.jms.StreamMessage对象，代表java原生对象的集合
• Map message: javax.jms.MapMessage对象，代表键值对

JMS Template应用

创建JMS Template

@Bean
public JmsTemplate createJMSTemplate() {

    SQSConnectionFactory sqsConnectionFactory = SQSConnectionFactory.builder()
            .withAWSCredentialsProvider(awsCredentialsProvider)
            .withEndpoint(endpoint)
            .withNumberOfMessagesToPrefetch(10).build();

    JmsTemplate jmsTemplate = new JmsTemplate(sqsConnectionFactory);
    jmsTemplate.setDefaultDestinationName(queueName);
    jmsTemplate.setDeliveryPersistent(false);
    return jmsTemplate;
}

发送消息

public void sendMessage(final String message) {

jmsTemplate.send(queueName, new MessageCreator() {
    @Override
    public Message createMessage(Session session) throws JMSException {
        return session.createTextMessage(message);
    }
});
}

同步接收消息

Message msg = jmsTemplate.receive("myQueue");

异步接收消息

异步处理消息， message listener container是必须的， 它用来绑定连接类工厂(connection factory), JMS Destination和message listener bean。 Spring有三种方式配置消息监听器： 实现javax.jms.MessageListener接口， 实现Spring的SessionAwareMessageListener(增加了对Jms Session的访问)和绑定一个POJO到Spring的MessageListenerAdapter类上。
对于消息监听器容器， Spring提供了两种： DefaultMessageListenerContainer和SimpleMessageListenerContainer， 两者都允许指定数量的并发监听线程 ，且DefaultMessageListenerContainer允许和XA Transaction集成， 对于使用本地事务管理器和不需要基于可变负载的线程、会话、连接调整的简单消息传递应用， 使用SimpleMessageListenerContainer, 对于使用外部事务管理器或XA事务的消息传递应用， 使用DefaultMessageListenerContainer。

@Bean
public DefaultMessageListenerContainer jmsListenerContainer() {

    SQSConnectionFactory sqsConnectionFactory = SQSConnectionFactory.builder()
            .withAWSCredentialsProvider(new DefaultAWSCredentialsProviderChain())
            .withEndpoint(endpoint)
            .withAWSCredentialsProvider(awsCredentialsProvider)
            .withNumberOfMessagesToPrefetch(10).build();

    DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
    dmlc.setConnectionFactory(sqsConnectionFactory);
    dmlc.setDestinationName(queueName);

    dmlc.setMessageListener(sqsListener);

    return dmlc;
}

实现MessageListener接口的监听器

@Component
public class SQSListener implements MessageListener {

private static final Logger LOGGER = LoggerFactory.getLogger(SQSListener.class);
public void onMessage(Message message) {
    TextMessage textMessage = (TextMessage) message;
    try {
        LOGGER.info("Received message "+ textMessage.getText());
    } catch (JMSException e) {
        LOGGER.error("Error processing message ",e);
    }
}
}

---

https://howtodoinjava.com/jms/jms-java-message-service-tutorial/
https://egkatzioura.com/2016/02/27/aws-sqs-and-spring-jms-integration/
https://blog.csdn.net/moonsheep_liu/article/details/6684948