JMS(Java Messaging Service)基础

编程入门行业动态更新时间:2024-10-26 01:26:14

JMS(Java Messaging Service)<a href=https://www.elefans.com/category/jswz/34/1770030.html style= 基础"/>

JMS(Java Messaging Service)基础

1.基础结构

1)图表结构

2)各部分介绍

ConnectionFactory

封装了一组连接配置参数，它已被定义为管理员。每个连接工厂都是队列(Queue)连接工厂或主题(Topic)连接工厂接口的一个实例

连接工厂的创建

private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);

关于DEFULT_BROKER_URL这个常量的来源，通过查找源码，可以得到

所以和tcp://localhost:61616得到的结果一样

注：

activeMQ默认配置下启动会启动8161和61616两个端口，其中8161是mq自带的管理后台的端口，61616是mq服务默认端口。

8161是后台管理系统，61616是给java用的tcp端口。

原文：(107条消息) activeMQ的两个默认端口8161和61616的区别_blueSkyGoGo的博客-CSDN博客_activemq默认端口号

用如下方式同样可以获取连接：

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");

Connection

使用连接来创建一个或多个会话

它有两种形式，即队列(Queue)连接或主题(Topic)连接

你可以用对应的连接工厂进行创建连接

QueueConnection queueConnection = queueConnectionFactory.createQueueConnection();
TopicConnection topicConnection = topicConnectionFactory.createTopicConnection();

建立连接后执行start()方法。在使用完毕后执行close()

Destination

客户端用于指定其生成的消息的目标和它所使用的消息的源的对象

Session

会话是用于生成和使用消息的单个线程上下文

会话的创建

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

第一个参数：是否支持事务，如果为true，则会忽略第二个参数，被jms服务器设置为SESSION_TRANSACTED
第一个参数为false时，第二个参数的值可为

Seesion.AUTO_ACKNOWLEDGE, //AUTO_ACKNOWLEDGE 为自动确认，客户端发送和接收消息不需要做额外的工作

Session.CLIENT_ACKNOWLEDGE,//CLIENT_ACKNOWLEDGE为客户端确认，客户端接收到消息后，必须调用javax.jms,Messagede1acknow方法

Session.DUPS_OK_ACKNOWLEDGE//DUPS_OK_ACKNOWLEDGE允许副本的确认模式，一旦接收方应用程序的方法调用从处理消息处返回，会话对象就会确认

您可以使用会话来创建消息生产者、消息使用者和消息。

MessageProducer producer = session.createProducer(queue);
MessageConsumer consumer = session.createConsumer(queue);

3)消息message

消息类型：

TextMessage: 一个java.lang。字符串对象（例如，可扩展标记语言文件的内容）
Object message : javax.jms.ObjectMessage，表示一个JAVA对象。
BytesMessage : javax.jms.BytesMessage，表示字节数据。
Stream message :javax.jms.StreamMessage，表示java原始值数据流。
MapMessage :一组名称/值对，其中名称作为字符串对象，值作为Java编程语言中的原始类型。

消息的发送：

Queue queue = session.createQueue(subject);
MessageProducer producer = session.createProducer(queue);
TextMessage message1 = session.createTextMessage("message1");
producer.send(message1);

消息的接收：

Message m = queueReceiver.receive();
if (m instanceof TextMessage) {
TextMessage message = (TextMessage)m;
Systems.out.println(“Reading message: “ + message.getText();
} else {
// Handle error
}

2.代码实现

1)发送方(producer,sender)

package com.activemq.sender;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
public class MessageSender {//URL of the JMS server. The default broker url is the localhost.private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;// a. Complete the queue nameprivate static String subject = "111";public static void main(String[] args) throws JMSException{// Getting JMS connection from the serverConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);// b. Complete the start connectionConnection connection = connectionFactory.createConnection();connection.start();// c. Create the sessionSession session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);Queue queue = session.createQueue(subject);//Destination. The queue will be created automatically on the serverMessageProducer producer = session.createProducer(queue);MessageProducer producer2 = session.createProducer(session.createQueue("queue"));TextMessage message1 = session.createTextMessage("111");producer.send(message1);producer2.send(message1);producer.close();session.close();connection.close();}
}

2)接收方(consumer,receiver)

package com.activemq.receiver;
import javax.jms.*;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
public class MessageReceiver {//URL of the JMS server. The default broker url is the localhost.private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;// a. Complete the queue nameprivate static String subject = "111";public static void main(String[] args) throws JMSException {// Getting JMS connection from the serverConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);// b. Complete the start connectionConnection connection = connectionFactory.createConnection();connection.start();Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);Destination destination = session.createQueue(subject);MessageConsumer consumer = session.createConsumer(destination);/*while (true){TextMessage textMessage = (TextMessage) consumer.receive();if (textMessage!=null){String text = textMessage.getText();System.out.println("The message received is "+text);}else {System.out.println("Message not received");break;}}*/if (consumer.receive() instanceof TextMessage textMessage) {System.out.println("Received message '" + textMessage.getText() + "'");consumer.close();session.close();connection.close();// c. Create the session//Destination. The queue will be created automatically on the server.}}
}