JMS入门简介

Posted 2022-02-18 liqipiao

一、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();
　　　　　　
　　　　　　
　　　　　　

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