码迷,mamicode.com
首页 > 其他好文 > 详细

JMS概述

时间:2018-07-14 23:02:27      阅读:282      评论:0      收藏:0      [点我收藏+]

标签:必须   nat   port   war   处理   apm   mic   throw   不同的   

介绍

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

JMS概述

标签:必须   nat   port   war   处理   apm   mic   throw   不同的   

原文地址:https://www.cnblogs.com/helloz/p/9311202.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!