JMS 教程

2024 年 8 月 29 日 | 4 分钟阅读

JMS (Java Message Service) 是一个 API，提供创建、发送和读取消息的功能。它提供松散耦合、可靠和异步的通信。

JMS 也被称为消息传递服务。

理解消息传递

消息传递是一种应用程序或软件组件进行通信的技术。

JMS 主要用于在一个应用程序向另一个应用程序发送和接收消息。

JMS 的需求

通常，用户向应用程序发送消息。但是，如果我们要从一个应用程序向另一个应用程序发送消息，我们需要使用 JMS API。

考虑一个场景，应用程序 A 在印度运行，而另一个应用程序 B 在美国运行。要从 A 应用程序向 B 应用程序发送消息，我们需要使用 JMS。

JMS 的优点

1) 异步： 要接收消息，客户端不需要发送请求。消息将自动到达客户端。

2) 可靠： 它保证消息被传递。

消息传递域

JMS 中有两种类型的消息传递域。

点对点消息传递域
发布/订阅消息传递域

1) 点对点 (PTP) 消息传递域

在 PTP 模型中，一条消息仅传递给一个接收者。在这里，队列用作面向消息的中间件 (MOM)。

队列负责保存消息，直到接收者准备好。

在 PTP 模型中，发送者和接收者之间没有时间依赖性。

2) 发布/订阅 (Pub/Sub) 消息传递域

在 Pub/Sub 模型中，一条消息传递给所有订阅者。这就像广播。在这里，主题用作面向消息的中间件，负责保存和传递消息。

在 PTP 模型中，发布者和订阅者之间存在时间依赖性。

JMS 编程模型

JMS 队列示例

要开发 JMS 队列示例，您需要安装任何应用程序服务器。在这里，我们使用 glassfish3 服务器，我们在其中创建两个 JNDI。

创建名为 myQueueConnectionFactory 的连接工厂
创建名为 myQueue 的目标资源

创建 JNDI 后，创建服务器和接收器应用程序。您需要在不同的控制台中运行服务器和接收器。在这里，我们使用 eclipse IDE，默认情况下它会在不同的控制台中打开。

1) 创建连接工厂和目标资源

通过 URL https://:4848 打开服务器管理控制台

使用用户名和密码登录。

单击 JMS 资源 -> 连接工厂 -> 新建，现在写入池名称并选择资源类型为 QueueConnectionFactory，然后单击确定按钮。

单击 JMS 资源 -> 目标资源 -> 新建，现在写入 JNDI 名称和物理目标名称，然后单击确定按钮。

2) 创建发送者和接收者应用程序

让我们看一下发送者和接收者的代码。请注意，接收器附加了侦听器，该侦听器将在用户发送消息时被调用。

文件：MySender.java

import java.io.BufferedReader;
import java.io.InputStreamReader;
import javax.naming.*;
import javax.jms.*;

public class MySender {
	public static void main(String[] args) {
		try
		{	//Create and start connection
			InitialContext ctx=new InitialContext();
			QueueConnectionFactory f=(QueueConnectionFactory)ctx.lookup("myQueueConnectionFactory");
			QueueConnection con=f.createQueueConnection();
			con.start();
			//2) create queue session
			QueueSession ses=con.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
			//3) get the Queue object
			Queue t=(Queue)ctx.lookup("myQueue");
			//4)create QueueSender object		
			QueueSender sender=ses.createSender(t);
			//5) create TextMessage object
			TextMessage msg=ses.createTextMessage();
			
			//6) write message
			BufferedReader b=new BufferedReader(new InputStreamReader(System.in));
			while(true)
			{
				System.out.println("Enter Msg, end to terminate:");
				String s=b.readLine();
				if (s.equals("end"))
					break;
				msg.setText(s);
				//7) send message
				sender.send(msg);
				System.out.println("Message successfully sent.");
			}
			//8) connection close
			con.close();
		}catch(Exception e){System.out.println(e);}
	}
}

文件：MyReceiver.java

import javax.jms.*;
import javax.naming.InitialContext;

public class MyReceiver {
	public static void main(String[] args) {
		try{
			//1) Create and start connection
			InitialContext ctx=new InitialContext();
			QueueConnectionFactory f=(QueueConnectionFactory)ctx.lookup("myQueueConnectionFactory");
			QueueConnection con=f.createQueueConnection();
			con.start();
			//2) create Queue session
			QueueSession ses=con.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
			//3) get the Queue object
			Queue t=(Queue)ctx.lookup("myQueue");
			//4)create QueueReceiver
			QueueReceiver receiver=ses.createReceiver(t);
			
			//5) create listener object
			MyListener listener=new MyListener();
			
			//6) register the listener object with receiver
			receiver.setMessageListener(listener);
			
			System.out.println("Receiver1 is ready, waiting for messages...");
			System.out.println("press Ctrl+c to shutdown...");
			while(true){				
				Thread.sleep(1000);
			}
		}catch(Exception e){System.out.println(e);}
	}

}

文件：MyListener.java

import javax.jms.*;
public class MyListener implements MessageListener {

	public void onMessage(Message m) {
		try{
		TextMessage msg=(TextMessage)m;
	
		System.out.println("following message is received:"+msg.getText());
		}catch(JMSException e){System.out.println(e);}
	}
}

首先运行接收者类，然后运行发送者类。

JMS 主题示例

它与 JMS 队列相同，但您需要将队列更改为主题，将发送者更改为发布者，将接收者更改为订阅者。

您需要创建 2 个名为 myTopicConnectionFactory 和 myTopic 的 JNDI。

文件：MySender.java

import java.io.BufferedReader;
import java.io.InputStreamReader;
import javax.naming.*;
import javax.jms.*;

public class MySender {
	public static void main(String[] args) {
		try
		{	//Create and start connection
			InitialContext ctx=new InitialContext();
			TopicConnectionFactory f=(TopicConnectionFactory)ctx.lookup("myTopicConnectionFactory");
			TopicConnection con=f.createTopicConnection();
			con.start();
			//2) create queue session
			TopicSession ses=con.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
			//3) get the Topic object
			Topic t=(Topic)ctx.lookup("myTopic");
			//4)create TopicPublisher object		
			TopicPublisher publisher=ses.createPublisher(t);
			//5) create TextMessage object
			TextMessage msg=ses.createTextMessage();
			
			//6) write message
			BufferedReader b=new BufferedReader(new InputStreamReader(System.in));
			while(true)
			{
				System.out.println("Enter Msg, end to terminate:");
				String s=b.readLine();
				if (s.equals("end"))
					break;
				msg.setText(s);
				//7) send message
				publisher.publish(msg);
				System.out.println("Message successfully sent.");
			}
			//8) connection close
			con.close();
		}catch(Exception e){System.out.println(e);}
	}
}

文件：MyReceiver.java

import javax.jms.*;
import javax.naming.InitialContext;

public class MyReceiver {
	public static void main(String[] args) {
		try	{
			//1) Create and start connection
			InitialContext ctx=new InitialContext();
			TopicConnectionFactory f=(TopicConnectionFactory)ctx.lookup("myTopicConnectionFactory");
			TopicConnection con=f.createTopicConnection();
			con.start();
			//2) create topic session
			TopicSession ses=con.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
			//3) get the Topic object
			Topic t=(Topic)ctx.lookup("myTopic");
			//4)create TopicSubscriber
			TopicSubscriber receiver=ses.createSubscriber(t);
			
			//5) create listener object
			MyListener listener=new MyListener();
			
			//6) register the listener object with subscriber
			receiver.setMessageListener(listener);
						
			System.out.println("Subscriber1 is ready, waiting for messages...");
			System.out.println("press Ctrl+c to shutdown...");
			while(true){				
				Thread.sleep(1000);
			}
		}catch(Exception e){System.out.println(e);}
	}

}

文件：MyListener.java

import javax.jms.*;
public class MyListener implements MessageListener {

	public void onMessage(Message m) {
		try{
		TextMessage msg=(TextMessage)m;
	
		System.out.println("following message is received:"+msg.getText());
		}catch(JMSException e){System.out.println(e);}
	}
}

下一个主题消息驱动 Bean