ok，了解消息在activemq中的传输流程后咱们来看看在哪里截取消息并发送到指定的消费者吧。java
一、使用git下载activemq源码
git clone https://git-wip-us.apache.org/repos/asf/activemq.git
二、git自动下载最新版本的源码，须要下载和你使用的mq程序对应的版本，上一步的网址中有tag一栏，选择对应的版本进去，查看对应的commit所产生的hashcode
git reset –hard <对应版本hashcode>
三、自定义消息到topic后推送给订阅者的消息分发策略
策略存放路径在项目根路径下的\activemq-broker\src\main\java\org\apache\activemq\broker\region\policy\
在此目录下随便选一个原有的*Policy.java文件，修改此文件内容
四、从新编译broker
在\activemq-broker\下执行mvn package进行打包（打包后的内容应该在target文件夹中）
把生成的activemq-broker-version.jar拷贝到使用的mq程序的lib目录下
五、修改activemq的配置文件node
接下来就是实现点对点的重点，自定义分发策略
一、点对点的实现是经过设置clientId来肯定两个点，由于MQTT框架没法设置message的properties属性，因此把clientId放到消息体中，用某个特殊符号隔开，我这里用的是@符号
二、producer.send(message);发送消息后到达topic中，若是此topic有订阅的消费者，那么才会进入到分发策略，而进入到哪一个分发策略是你在第五步配置的\conf\activemq.xml中指定的那个，配置文件更改后就会进入到下面的策略中。
三、分发策略中定义，必须是在名为.ptp结尾的topic中才会进行点对点推送，否则就和正常的同样一对多的推送git
String msgBody = "message body";
String clientId = "mqtt-1012";
producer.send(clientId+"@"+msgBody);
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;
public class PriorityDispatchPolicy extends SimpleDispatchPolicy {
public boolean dispatch(MessageReference node, MessageEvaluationContext msgContext, List<Subscription> consumers) throws Exception {
//指定特定后缀名的topic进入自定义分发策略
if(!node.getMessage().getDestination().getQualifiedName().endsWith(".ptp"))
return super.dispatch(node, msgContext, consumers);
String _clientId = "";
//获取消息内容
ByteSequence sequence = node.getMessage().getContent();
byte[] message = sequence.getData();
String strMessage = new String(message, "utf-8");
//format：clientID@messageBody
if (!"".equals(strMessage) && strMessage.indexOf("@")>-1)
_clientId = strMessage.substring(0, strMessage.indexOf("@"));
if ("".equals(_clientId))
return super.dispatch(node, msgContext, consumers);
strMessage = strMessage.substring(strMessage.indexOf("@") + 1, strMessage.length());
node.getMessage().setContent(new ByteSequence(strMessage.getBytes("utf-8")));
ActiveMQDestination _destination = node.getMessage().getDestination();
int count = 0;
for (Subscription sub : consumers) {
// Don't deliver to browsers
if (sub.getConsumerInfo().isBrowser()) {
continue;
}
// Only dispatch to interested subscriptions
if (!sub.matches(node, msgContext)) {
sub.unmatched(node);
continue;
}
if (_clientId != null && _destination.isTopic() && _clientId.equals(sub.getContext().getClientId())
&& _destination.getQualifiedName().endsWith(".ptp")) {
//把消息推送给知足要求的subscription
sub.add(node);
count++;
} else {
sub.unmatched(node);
}
}
return count > 0;
}
}
下面展现的是第五步修改的内容，用文本编辑器打开mq程序根目录\conf\activemq.xml文件，寻找到broker标签，修改其中的内容web
<destinationPolicy>
<policyMap>
<policyEntries>
<policyEntry topic=">" >
<dispatchPolicy>
<!-- 这里的priorityDispatchPolicy是你选对的策略分发类，首字母小写 -->
<priorityDispatchPolicy />
</dispatchPolicy>
<pendingMessageLimitStrategy>
<constantPendingMessageLimitStrategy limit="1000"/>
</pendingMessageLimitStrategy>
</policyEntry>
</policyEntries>
</policyMap>
</destinationPolicy>
直接贴代码，不作解释啦apache
public class P2PMqttTest {
public static Connection connection;
public static Statement stat;
public static void main(String[] args) throws JMSException {
DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("admin", "admin", "tcp://localhost:61616");
//consumer
subMsg(factory, "mqtt-1011");
subMsg(factory, "mqtt-1012");
subMsg(factory, "aaaaaa");
//product
sendMsg(factory);
}
private static void subMsg(ActiveMQConnectionFactory factory, String clientId) throws JMSException {
Connection connection = factory.createConnection();
connection.setClientID(clientId);
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Topic topic = session.createTopic("channel.ptp");
MessageConsumer consumer = session.createConsumer(topic);
consumer.setMessageListener(new ConsumeListenerMessage(clientId));
}
private static void sendMsg(ActiveMQConnectionFactory factory) throws JMSException {
Connection connection = factory.createConnection();
connection.setClientID("mqtt-1022");
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Topic topic = session.createTopic("channel.ptp");
MessageProducer producer = session.createProducer(topic);
//clientId@messageBody
Message message = session.createTextMessage("mqtt-1011"+"@"+"test");
producer.send(message);
}
static class ConsumeListenerMessage implements MessageListener {
private String thisClientId;
public void onMessage(Message message) {
TextMessage msg = (TextMessage) message;
try {
String clientId = msg.getStringProperty("PTP_CLIENTID");
String msgBody = msg.getText();
System.out.println("============================================");
System.out.println(thisClientId+" recive message");
System.out.println(msgBody);
System.out.println("============================================");
} catch (JMSException e) {
e.printStackTrace();
}
}
public ConsumeListenerMessage(String thisClientId) {
this.thisClientId = thisClientId;
}
}
}