activemq结合mqtt发送p2p消息

╰半夏微凉° 2022-06-10 08:21 377阅读 0赞

**实现思路**：所有用户订阅一个主题，当服务器端发起推送时使用jms协议发送消息到主题，并设置附带属性为目标用户的clientId,对该主题进行自定义分发策略

**1.下载mqtt源码**

自行下载，本案例以5.5.10为例

**2.自定义分发策略**

添加一个分发策略带指定的源码包路径：org.apache.activemq.broker.region.policy

注：一定要放在此包下面

*以下为完整的class内容：*

package org.apache.activemq.broker.region.policy;

import java.util.List;

import org.apache.activemq.broker.region.MessageReference;

import org.apache.activemq.broker.region.Subscription;

import org.apache.activemq.command.ActiveMQDestination;

import org.apache.activemq.filter.MessageEvaluationContext;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

/\*\*

\* ClientIdFilterDispatchPolicy dispatches messages in a topic to a given

\* client. Then the message with a "PTP\_CLIENTID" property, can be received by a

\* mqtt client with the same clientId.

\* *@org.apache.xbean.XBean*

\*/

public class ClientIdFilterDispatchPolicy extends SimpleDispatchPolicy \{

private static final Log LOG = LogFactory.getLog(ClientIdFilterDispatchPolicy.class);

public static final String PTP\_CLIENTID = "PTP\_CLIENTID";

//可自定义消息目标id在消息属性中的key

private String ptpClientId = PTP\_CLIENTID;

public boolean dispatch(MessageReference node, MessageEvaluationContext msgContext, List<Subscription> consumers)

throws Exception \{

if (LOG.isInfoEnabled()) \{

LOG.info("===============Enter ClientIdFilterDispatchPolicy........");

// 获取消息中的目标 客户端id

Object clientId = node.getMessage().getProperty(ptpClientId);

// 如果没有，直接广播

if (clientId == null) \{

return super.dispatch(node, msgContext, consumers);

if (LOG.isInfoEnabled()) \{

LOG.info("===============Client id : " + clientId);

// 获取当前消息类型，此处主要是限制为主题模式

ActiveMQDestination destination = node.getMessage().getDestination();

int count = 0;

// 遍历所有订阅者

for (Subscription sub : consumers) \{

if (LOG.isInfoEnabled()) \{

LOG.info("===============consumers id: " + sub.getContext().getClientId());

// 不交于浏览器

if (sub.getConsumerInfo().isBrowser()) \{

continue;

// 只发送给感兴趣的订阅

if (!sub.matches(node, msgContext)) \{

sub.unmatched(node);

continue;

if (LOG.isInfoEnabled()) \{

LOG.info("==============destination clientId : " + clientId);

// 消息中带有的目标id不为空，也为主题模式，并且当前的消费者的id和消息中的目标id相同，则投递消息

if (clientId != null && destination.isTopic() && clientId.equals(sub.getContext().getClientId())) \{

if (LOG.isInfoEnabled()) \{

LOG.info("==============Send p2p message to : " + clientId);

LOG.info("==============Top ic : " + destination.isTopic());

sub.add(node);

count++;

\} else \{

// 过滤消息，不进行投递

LOG.info("==============Un consumers subscription!");

sub.unmatched(node);

return count > 0;

public String getPtpClientId() \{

return ptpClientId;

public void setPtpClientId(String ptpClientId) \{

this.ptpClientId = ptpClientId;

*红色tips:*

*在类开头的注释中需要加 ：@org.apache.xbean.XBean，否则编译的时候无法加载到activemq-spring/actimvemq.xsd文件内*

3.maven编译打包以及替换

maven项目顶层目录执行：mvn -Dtest=false clean install

替换 activemq-spring-x.x.x.jar和activemq-broker-x.x.x-.jar两个文件

*tips：*

*1)文件目录：* *assembly/target 解压apache-activemq-version-bin.zip文件，进入解压后的目录/lib 文件夹中*

*2)代码编写自定义的policy后，重新编译，acitvemq会自动将自定义好的policy以首字母小写的方式自动生产到activemq.xsd文件中，如果该文件没有你自定义的policy定义，则再activemq.xml文件中引用的时候会报错。*

*activemq.xsd文件是编译打包后生成的，文件路径：activemq-spring-version.jar中，解压后会看到*

![Image 1][]

*然后可以再里面找到自定义的policy的定义。*

![Image 1][]

**4.activemq.xml文件中配置自己写的分发策略：**

</dispatchPolicy>

</policyEntry>

如下图所示，表示PTP开头的所有主题都会进行该自定义分发策略

![Image 1][]

*tips:如果自定义的policy没有在指定的包目录，或者类名包含了数字，或者其他原因导致没有生成到* *activemq.xsd* *文件中，那么此步骤配置改项启动的时候会出现错误*

**5.消息发送和接收案例：**

*1）服务端使用jms发送消息：*

import java.util.Scanner;

import javax.jms.Connection;

import javax.jms.DeliveryMode;

import javax.jms.JMSException;

import javax.jms.MessageProducer;

import javax.jms.Session;

import javax.jms.TextMessage;

import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnectionFactory;

import org.apache.commons.lang3.StringUtils;

/\*\*

\* \[简要描述\]:

\* \[详细描述\]:

\* @author linlinxiao

\* @version 1.0, 2017年7月21日

\* @since V100R001C10

\*/

public class TopicProducerTest

public static void main(String\[\] args) throws JMSException

// 创建连接工厂

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

String PTP\_CLIENTID = "PTP\_CLIENTID";

// 鉴权，如没有开启可省略

factory.setUserName("admin");

factory.setPassword("admin123");

// 创建JMS连接实例，并启动连接

Connection connection = factory.createConnection();

connection.start();

// 创建Session对象，不开启事务

Session session = connection.createSession(false, Session.AUTO\_ACKNOWLEDGE);

// 创建主题

Topic topic = session.createTopic("PTP.test");

// 创建生成者

MessageProducer producer = session.createProducer(topic);

// 设置消息不需持久化。默认消息需要持久化

producer.setDeliveryMode(DeliveryMode.NON\_PERSISTENT);

Scanner sc = new Scanner(System.in);

boolean isStart = true;

String userMsg = "";

String msg = "";

TextMessage message = null;

String\[\] messages = null;

String clientId = null;

while (isStart)

userMsg = sc.nextLine();

if (StringUtils.isBlank(userMsg) || "stop".equals(userMsg))

System.out.println("Stop producer message!");

isStart = false;

messages = userMsg.split(":");

msg = "Hello MQ,Client msg:" + messages\[0\];

message = session.createTextMessage(msg);

if (messages.length == 2)

clientId = messages\[1\];

// 发送指定消息，配合主题分发策略使用，以附带用户ID ，分发策略对特定的主题进行拦截解析分发

if (StringUtils.isNotBlank(clientId))

message.setStringProperty(PTP\_CLIENTID, clientId);

// 发送消息。non-persistent 默认异步发送；persistent 默认同步发送

producer.send(message);

sc.close();

// 关闭连接

producer.close();

session.close();

connection.close();

2）mqtt端接收使用

import org.eclipse.paho.client.mqttv3.MqttClient;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;

import org.eclipse.paho.client.mqttv3.MqttException;

import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/\*\*

\* \[简要描述\]:

\* \[详细描述\]:

\* @author xiaolinlin

\* @version 1.0, 2017年6月8日

\* @since V100R001C00

\*/

public class MqttTestClient

public static final String HOST = "tcp://localhost:1883";

public static final String TOPIC = "test";

private String clientId;

private MqttClient client;

private MqttConnectOptions options;

private String userName = "admin";

private String passWord = "admin123";

public MqttTestClient(String clientId)

this.clientId = clientId;

private void start() throws MqttException

try

// host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存

client = new MqttClient(HOST, clientId, new MemoryPersistence());

// MQTT的连接设置

options = new MqttConnectOptions();

// 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接

options.setCleanSession(true);

// 设置连接的用户名

options.setUserName(userName);

// 设置连接的密码

options.setPassword(passWord.toCharArray());

// 设置超时时间 单位为秒

options.setConnectionTimeout(10);

// 设置会话心跳时间 单位为秒 服务器会每隔1.5\*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制

options.setKeepAliveInterval(20);

// // 设置回调

// MqttTopic topic = client.getTopic(TOPIC+"/test/");

// // setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息

// options.setWill(topic, "close".getBytes(), 2, true);

client.connect(options);

client.setCallback(new ClientCallback(client, options));

// 订阅消息

int\[\] Qos = \{1\};

String\[\] topic1 = \{"PTP/test"\};

client.subscribe(topic1, Qos);

catch (Exception e)

e.printStackTrace();

public static void main(String\[\] args) throws MqttException

String clientId = "admin1";

MqttTestClient client = new MqttTestClient(clientId);

client.start();

callback类：

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;

import org.eclipse.paho.client.mqttv3.MqttCallback;

import org.eclipse.paho.client.mqttv3.MqttClient;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;

import org.eclipse.paho.client.mqttv3.MqttException;

import org.eclipse.paho.client.mqttv3.MqttMessage;

import org.eclipse.paho.client.mqttv3.MqttSecurityException;

/\*\*

\* \[简要描述\]:

\* \[详细描述\]:

\* @author xiaolinlin

\* @version 1.0, 2017年6月8日

\* @since V100R001C00

\*/

public class ClientCallback implements MqttCallback

private MqttClient client;

private MqttConnectOptions options;

public ClientCallback(MqttClient client,MqttConnectOptions options) \{

this.client = client;

this.options = options;

@Override

public void connectionLost(Throwable cause)

// 连接丢失后，一般在这里面进行重连

System.out.println("连接断开，可以做重连");

try

client.connect(options);

catch (MqttSecurityException e)

// TODO Auto-generated catch block

e.printStackTrace();

catch (MqttException e)

// TODO Auto-generated catch block

e.printStackTrace();

// while(!sampleClient.isConnected())\{

// try \{

// Thread.sleep(1000);

// sampleClient.connect(connOpts);

// //客户端每次上线都必须上传自己所有涉及的订阅关系，否则可能会导致消息接收延迟

// sampleClient.subscribe(topicFilters,qos);

// \} catch (Exception e) \{

// e.printStackTrace();

// \}

@Override

public void messageArrived(String topic, MqttMessage message) throws Exception

// subscribe后得到的消息会执行到这里面

System.out.println("接收消息主题 : " + topic);

System.out.println("接收消息Qos : " + message.getQos());

String msg = new String(message.getPayload());

System.out.println("接收消息到服务端内容 : " + msg);

if(msg.contains("close")) \{

client.close();

@Override

public void deliveryComplete(IMqttDeliveryToken token)

System.out.println("deliveryComplete---------" + token.isComplete());

*红色tips：*

*在jms发送消息端主题格式为 P2P.smile,在mqtt客户端订阅使用的主题格式为P2P/smile。也就是说mqtt中和jms中使用的分隔符是不一样的。这个坑很难察觉，容易误导配置的策略不生效。*

**6.demo演示部分效果图：**

1）以控制台模式启动的activemq 观看日志结果：

![Image 1][]

会看到日志记录已经进入了我们自定义的策略

其他自行验证

7.参考博文：https://blog.csdn.net/kimmking/article/details/17449019

[Image 1]:

activemq结合mqtt发送p2p消息

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 ActiveMQ之P2P点对点通信方式

相关 P2P技术基础

相关 P2P原理

相关 activemq结合mqtt发送p2p消息

相关 ActiveMQ发送queue消息

相关 Spring2.5整合ActiveMQ 5.2（P2P文本消息）

相关 P2P技术介绍

相关 P2P协议入门

相关初识activemq消息中间件(p2p模式)

相关 P2P协议简介