介绍
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消息组成
-
Message Header
包含一系列属性被服务端和客户端用来发布及识别消息， 这些属性包括：
– JMSDestination
– JMSDeliveryMode
– JMSMessageID
– JMSTimestamp
– JMSCorrelationID
– JMSReplyTo
– JMSRedelivered
– JMSType
– JMSExpiration
– JMSPriority -
Message Properties
应用程序可以创建，设置消息属性，这些属性是一系列键值对。 通过读取消息属性可以过滤消息， JMS也提供了一些预定义的服务端可以识别的消息属性， 这些属性是可选的。 -
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