第 11 章 截获消息
使用 AMQ Broker，您可以拦截进入或退出代理的数据包，允许您审核数据包或过滤信息。拦截器可以更改它们拦截的数据包，这使它们的功能强大，但也具有危险性。
您可以开发拦截器来满足您的业务需求。拦截器是特定于协议的，必须实施相应的接口。
拦截器必须实施 intercept （） 方法，这将返回布尔值。如果值为 true，则消息数据包将继续。如果为 false，则进程将被中止，不会调用其他拦截器，并且不会进一步处理消息数据包。
11.1. 创建 Interceptors
您可以创建自己的传入和传出拦截器。所有拦截器都是特定的协议，对于分别进入或退出服务器的任何数据包调用。这可让您创建拦截器来满足审计数据包等业务要求。拦截器可以更改它们拦截的数据包。这使得它们的功能强大以及潜在的危险性，因此务必谨慎使用。
拦截器及其依赖项必须放在代理的 Java 类路径中。由于 <broker_instance_dir> /lib 目录默认是类路径的一部分，因此可以使用 & lt;broker_instance_dir> /lib 目录。
流程
以下示例演示了如何创建一个拦截器来检查传递给它的每个数据包的大小。请注意，这些示例为每个协议实施特定的接口。
实施适当的接口并覆盖其 intercept
（）方法。如果您使用 AMQP 协议，实施
org.apache.activemq.artemis.protocol.amqp.broker.AmqpInterceptor接口。package com.example; import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessage; import org.apache.activemq.artemis.protocol.amqp.broker.AmqpInterceptor; import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; public class MyInterceptor implements AmqpInterceptor { private final int ACCEPTABLE_SIZE = 1024; @Override public boolean intercept(final AMQPMessage message, RemotingConnection connection) { int size = message.getEncodeSize(); if (size <= ACCEPTABLE_SIZE) { System.out.println("This AMQPMessage has an acceptable size."); return true; } return false; } }如果您使用的是核心协议，您的拦截器必须实现
org.apache.artemis.activemq.api.core.Interceptor接口。package com.example; import org.apache.artemis.activemq.api.core.Interceptor; import org.apache.activemq.artemis.core.protocol.core.Packet; import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; public class MyInterceptor implements Interceptor { private final int ACCEPTABLE_SIZE = 1024; @Override boolean intercept(Packet packet, RemotingConnection connection) throws ActiveMQException { int size = packet.getPacketSize(); if (size <= ACCEPTABLE_SIZE) { System.out.println("This Packet has an acceptable size."); return true; } return false; } }如果您使用的是 MQTT 协议，请实施
org.apache.activemq.artemis.core.protocol.mqtt.MQTTInterceptor接口。package com.example; import org.apache.activemq.artemis.core.protocol.mqtt.MQTTInterceptor; import io.netty.handler.codec.mqtt.MqttMessage; import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; public class MyInterceptor implements Interceptor { private final int ACCEPTABLE_SIZE = 1024; @Override boolean intercept(MqttMessage mqttMessage, RemotingConnection connection) throws ActiveMQException { byte[] msg = (mqttMessage.toString()).getBytes(); int size = msg.length; if (size <= ACCEPTABLE_SIZE) { System.out.println("This MqttMessage has an acceptable size."); return true; } return false; } }如果您使用 STOMP 协议，请实施
org.apache.activemq.artemis.core.protocol.stomp.StompFrameInterceptor接口。package com.example; import org.apache.activemq.artemis.core.protocol.stomp.StompFrameInterceptor; import org.apache.activemq.artemis.core.protocol.stomp.StompFrame; import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; public class MyInterceptor implements Interceptor { private final int ACCEPTABLE_SIZE = 1024; @Override boolean intercept(StompFrame stompFrame, RemotingConnection connection) throws ActiveMQException { int size = stompFrame.getEncodedSize(); if (size <= ACCEPTABLE_SIZE) { System.out.println("This StompFrame has an acceptable size."); return true; } return false; } }
