JMS入门简介

Posted liqipiao

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了JMS入门简介相关的知识,希望对你有一定的参考价值。

一、JMS是什么

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

2.JMS是一种与厂商无关的 API,用来访问收发系统消息,它类似于JDBC(Java Database Connectivity)。这里,JDBC 是可以用来访问许多不同关系数据库的 API,而 JMS 则提供同样与厂商无关的访问方法,以访问消息收发服务。

二、对象模型

1)连接工厂。连接工厂(ConnectionFactory)是由管理员创建,并绑定到JNDI树中。客户端使用JNDI查找连接工厂,然后利用连接工厂创建一个JMS连接。
2)JMS连接。JMS连接(Connection)表示JMS客户端和服务器端之间的一个活动的连接,是由客户端通过调用连接工厂的方法建立的。
3)JMS会话。JMS会话(Session)表示JMS客户与JMS服务器之间的会话状态。JMS会话建立在JMS连接上,表示客户与服务器之间的一个会话线程。
4)JMS目的。JMS目的(Destination),又称为消息队列,是实际的消息源。
5)JMS生产者和消费者。生产者(Message Producer)和消费者(Message Consumer)对象由Session对象创建,用于发送和接收消息。
6)JMS消息通常有两种类型:
  ① 点对点(Point-to-Point)。在点对点的消息系统中,消息分发给一个单独的使用者。点对点消息往往与队列(javax.jms.Queue)相关联。
  ② 发布/订阅(Publish/Subscribe)。发布/订阅消息系统支持一个事件驱动模型,消息生产者和消费者都参与消息的传递。生产者发布事件,而使用者订阅感兴趣的事件,并使用事件。该类型消息一般与特定的主题(javax.jms.Topic)关联。

三、两种通用类型特点

点对点消息模型有如下特性:

      1.每个消息只有一个接受者

    2.消息发送者和消息接受者并没有时间依赖性

       3.当消息发送者发送消息的时候,无论接收者程序在不在运行,都能获取到消息

发布/订阅消息模型有如下特性:

       1.一个消息可以传递给多个订阅者

       2.发布者和订阅者有时间依赖性,只有当客户端创建订阅后才能接受消息,且订阅者需一直保持活动状态以接收消息

  技术图片

四、上代码

  1.点对点模式:

    1).消息的生产者      

      package com.jinchi.jms;

      import javax.jms.Connection;
      import javax.jms.ConnectionFactory;
      import javax.jms.Destination;
      import javax.jms.JMSException;
      import javax.jms.MessageProducer;
      import javax.jms.Session;
      import javax.jms.TextMessage;

      import org.apache.activemq.ActiveMQConnection;
      import org.apache.activemq.ActiveMQConnectionFactory;

      public class JMSProducer
      private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名
      private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码
      private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址
      private static final int SENDNUM=10; // 发送的消息数量
      /**
      * 消息生产者
      * @param args
      */
      public static void main(String[] args)
      //连接工厂
      ConnectionFactory cFactory;
      //连接
      Connection connection = null;
      //会话
      Session session;
      //消息目的地
      Destination destination;
      //消息生产者
      MessageProducer messageProducer;
      //实例化工厂
      cFactory=new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEURL);
      try
      //通过工厂获得连接
      connection=cFactory.createConnection();
      //启动连接
      connection.start();
      //创建session
      session=connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
      //创建消息队列
      destination=session.createQueue("FirstQueue1");
      //创建消息生产者
      messageProducer=session.createProducer(destination);
      //发送消息
      sendMessage(session,messageProducer);
      session.commit();
       catch (Exception e)
      e.printStackTrace();
      finally
      if (connection!=null)
      try
      connection.close();
       catch (JMSException e)
      e.printStackTrace();
      
      
      
      
      public static void sendMessage(Session session, MessageProducer messageProducer)throws Exception
      for (int i = 0; i < SENDNUM; i++)
      TextMessage textMessage=session.createTextMessage("ActiveMQ发送次数:"+i);
      System.out.println("发送消息:"+"ActiveMQ发送次数:"+i);
      messageProducer.send(textMessage);
      
      

      

    2).消息的消费者  

      package com.jinchi.jms;

      import javax.jms.Connection;
      import javax.jms.ConnectionFactory;
      import javax.jms.Destination;
      import javax.jms.JMSException;
      import javax.jms.MessageConsumer;
      import javax.jms.Session;

      import org.apache.activemq.ActiveMQConnection;
      import org.apache.activemq.ActiveMQConnectionFactory;

      public class JMSconsumer2
      private static final String USERNAME=ActiveMQConnection.DEFAULT_USER;
      private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD;
      private static final String BORK_STRING=ActiveMQConnection.DEFAULT_BROKER_URL;
      public static void main(String[] args)
      //连接工厂
      ConnectionFactory cFactory;
      //连接
      Connection connection;
      //会话
      Session session;
      //消费的目的地
      Destination destination;
      //消息的消费者
      MessageConsumer messageConsumer;
      //实例化工厂
      cFactory=new ActiveMQConnectionFactory(USERNAME,PASSWORD,BORK_STRING);
      try
      //通过工厂获取连接对象
      connection=cFactory.createConnection();
      //启动连接
      connection.start();
      //创建会话
      session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
      //创建连接的消息队列
      destination=session.createQueue("FirstQueue1");
      //创建消息消费者
      messageConsumer=session.createConsumer(destination);
      //注册消息监听
      messageConsumer.setMessageListener(new JMSListenre());
       catch (JMSException e)
      // TODO Auto-generated catch block
      e.printStackTrace();
      
      

      

    3).消息消费者监听事件   

      package com.jinchi.jms;

      import javax.jms.JMSException;
      import javax.jms.Message;
      import javax.jms.MessageListener;
      import javax.jms.TextMessage;

      /**
      * 消息监听
      * @author Admin
      *
      */
      public class JMSListenre implements MessageListener

      @Override
      public void onMessage(Message message)
      try
      System.out.println("收到的消息:"+((TextMessage)message).getText());
       catch (JMSException e)
      // TODO Auto-generated catch block
      e.printStackTrace();
      
      

      

 

  2.发布/订阅消息模型:

    1).消息的发布者

      package com.jinchi.jms2;

      import javax.jms.Connection;
      import javax.jms.ConnectionFactory;
      import javax.jms.Destination;
      import javax.jms.JMSException;
      import javax.jms.MessageProducer;
      import javax.jms.Session;
      import javax.jms.TextMessage;

      import org.apache.activemq.ActiveMQConnection;
      import org.apache.activemq.ActiveMQConnectionFactory;

      public class JMSProducer
      private static final String USERNAME=ActiveMQConnection.DEFAULT_USER;
      private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD;
      private static final String BORK=ActiveMQConnection.DEFAULT_BROKER_URL;
      private static final int SEND=10;
      public static void main(String[] args)
      // 连接工厂
      ConnectionFactory cFactory;
      //连接
      Connection connection = null;
      //会话
      Session session;
      //消息的目的地
      Destination destination;
      //消息的生产者
      MessageProducer messageProducer;
      //实例化工厂
      cFactory=new ActiveMQConnectionFactory(USERNAME,PASSWORD,BORK);
      try
      //通过工厂获取连接对象
      connection=cFactory.createConnection();
      //连接
      connection.start();
      //会话
      session=connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
      //创建消息队列
      destination=session.createTopic("FirstQueue2");
      //创建消息生产者
      messageProducer=session.createProducer(destination);
      //发送消息
      sendMessage(session,messageProducer);
      session.commit();
       catch (JMSException e)
      // TODO Auto-generated catch block
      e.printStackTrace();
      finally
      if (connection!=null)
      try
      connection.close();
       catch (JMSException e)
      // TODO Auto-generated catch block
      e.printStackTrace();
      
      
      

      
      public static void sendMessage(Session session, MessageProducer messageProducer)
      for (int i = 0; i < SEND; i++)
      try
      TextMessage textMessage=session.createTextMessage("ActiveMQ发送次数:"+i);
      System.out.println("发送消息:"+"ActiveMQ发送次数:"+i);
      //发送
      messageProducer.send(textMessage);
       catch (JMSException e)
      // TODO Auto-generated catch block
      e.printStackTrace();
      
      
      

      

    2).消息的订阅者,需要创建两个订阅者,用于查看发布的消息是否都能被两个订阅者消费

      ①第一个订阅者:    

      package com.jinchi.jms2;

      import javax.jms.Connection;
      import javax.jms.ConnectionFactory;
      import javax.jms.Destination;
      import javax.jms.JMSException;
      import javax.jms.MessageConsumer;
      import javax.jms.Session;

      import org.apache.activemq.ActiveMQConnection;
      import org.apache.activemq.ActiveMQConnectionFactory;

      public class JMSconsumer
      private static final String USERNAME=ActiveMQConnection.DEFAULT_USER;
      private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD;
      private static final String BORKURL=ActiveMQConnection.DEFAULT_BROKER_URL;

      public static void main(String[] args)
      //连接工厂
      ConnectionFactory cFactory;
      //连接
      Connection connection;
      //会话
      Session session;
      //消费目的地
      Destination destination;
      //消息消费者
      MessageConsumer messageConsumer;
      //实例化工厂
      cFactory =new ActiveMQConnectionFactory(USERNAME,PASSWORD,BORKURL);
      try
      //通过工厂获取连接对象
      connection=cFactory.createConnection();
      //启动
      connection.start();
      //会话
      session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
      //消费目的地
      destination=session.createTopic("FirstQueue2");
      //创建消息消费者
      messageConsumer=session.createConsumer(destination);
      //创建消费者监听
      messageConsumer.setMessageListener(new JMSListener1());
       catch (JMSException e)
      e.printStackTrace();
      

      

      

      第一个订阅者监听事件:  

      package com.jinchi.jms2;

      import javax.jms.JMSException;
      import javax.jms.Message;
      import javax.jms.MessageListener;
      import javax.jms.TextMessage;

      public class JMSListener1 implements MessageListener

      @Override
      public void onMessage(Message message)
      try
      System.out.println("订阅者一收到的消息:"+((TextMessage)message).getText());
       catch (JMSException e)
      // TODO Auto-generated catch block
      e.printStackTrace();
      

       

      

      ②第二个订阅者:    

      package com.jinchi.jms2;

      import javax.jms.Connection;
      import javax.jms.ConnectionFactory;
      import javax.jms.Destination;
      import javax.jms.JMSException;
      import javax.jms.MessageConsumer;
      import javax.jms.Session;

      import org.apache.activemq.ActiveMQConnection;
      import org.apache.activemq.ActiveMQConnectionFactory;

      public class JMSconsumer2
      private static final String USERNAME=ActiveMQConnection.DEFAULT_USER;
      private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD;
      private static final String BORKURL=ActiveMQConnection.DEFAULT_BROKER_URL;

      public static void main(String[] args)
      //连接工厂
      ConnectionFactory cFactory;
      //连接
      Connection connection;
      //会话
      Session session;
      //消息目的地
      Destination destination;
      //消息消费者
      MessageConsumer messageConsumer;
      //实例化工厂
      cFactory=new ActiveMQConnectionFactory(USERNAME,PASSWORD,BORKURL);
      try
      //通过工厂获取连接对象
      connection=cFactory.createConnection();
      //连接
      connection.start();
      //会话
      session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
      //消费的目的地
      destination=session.createTopic("FirstQueue2");
      //创建消息消费者
      messageConsumer=session.createConsumer(destination);
      //创建消息监听
      messageConsumer.setMessageListener(new JMSListener2());
       catch (JMSException e)
      // TODO Auto-generated catch block
      e.printStackTrace();
      

      

      

      第二个订阅者监听事件:   

      package com.jinchi.jms2;

      import javax.jms.JMSException;
      import javax.jms.Message;
      import javax.jms.MessageListener;
      import javax.jms.TextMessage;

      public class JMSListener2 implements MessageListener

      @Override
      public void onMessage(Message message)
      try
      System.out.println("订阅者二收到的消息:"+((TextMessage)message).getText());
       catch (JMSException e)
      // TODO Auto-generated catch block
      e.printStackTrace();
      
      
      

  测试发布/订阅消息模型时,先启动两个订阅者,然后再启动发布者,否则无法达到真正的测试。

      



 

以上是关于JMS入门简介的主要内容,如果未能解决你的问题,请参考以下文章

kafka入门:简介使用场景设计原理主要配置及集群搭建

零基础快速入门SpringBoot2.0教程

ActiveMQ入门介绍

RabbitMQ入门

Rabbitmq入门

JMS好文