实现基于订阅发布的消息总线
- 往期链接
- 实现的功能及形式
- 功能设计及代码实现
- 生产者
- 使用redis实现生产者
- 消费者
- 使用redis实现消费者
- 配套@Subscribe注解
- 实现BeanPostProcessor监听所有的bean创建
- EventBus对象
往期链接
从零开始搭建属于自己的物联网平台（一）需求分析以及架构设计
实现的功能及形式
- 以stater方式实现
- 支持市面上常见的发布订阅的三方组件，比如各种MQ，redis stream等
- 提供注解，快速实现订阅，同样支持代码来动态添加订阅
功能设计及代码实现
首先，像这种平台的功能，一定要做好抽象设计，预留好功能扩展的接口
处理流程图
这里生产者处理逻辑比较简单，当设备消息或者其他业务的消息产生的时候，调用生产者实现的发布方法将消息推送至消息总线。
消费者这里处理就稍微复杂一些，首先在消费者服务启动时需要先扫描所有的服务内的订阅者（一个@Subscribe修饰的方法就是一个订阅者，为了防止混淆这里将这些订阅者称为MessageAdapter），将这些Adapter进行注册，并根据订阅的topic做好分组（有可能会有多个Adapter订阅同一个topic），消费者持续监听消息总线,如果有是多个实例那么做好分组，当订阅的消息到来之后，首先做消息的解析，解析出内部消息的topic,再从内存中取得对应的Adapter做服务内部的消息多播（这里其实就是仿照的spring event的处理逻辑），Adapter接收到多播消息之后再进行对应的业务处理。
生产者
定义生产者抽象接口，以后兼容的第三方组件只要实现该接口就好。
/**
* 消息总线代码发布
*/
public interface MessagePublisher {
/**
* 消息发布
*
* @param message 消息体
* @return 发布结果
*/
Boolean publish(AdapterMessage message);
}
使用redis实现生产者
这里使用@ConditionalOnProperty来修饰该实现类，这样目的是让redis stream作为消息总线的默认实现，如果后续添加别的实现，可以在配置文件中便利的配置。
/**
* 消息总线发布Redis实现
*/
@Component
@ConditionalOnProperty(prefix = "stream.broker", name = "type", havingValue = "redis", matchIfMissing = true)
public class RedisMessagePublisher implements MessagePublisher {
@Value("${stream.topic:/stream}")
String topic;
private final RedisConnectionFactory connectionFactory;
public RedisMessagePublisher(RedisConnectionFactory connectionFactory) {
this.connectionFactory = connectionFactory;
}
@Override
public Boolean publish(AdapterMessage message) {
Map value = new HashMap();
value.put("topic".getBytes(), message.getTopic().getBytes());
value.put("payload".getBytes(), message.getPayload().getBytes());
ByteRecord byteRecord = StreamRecords.rawBytes(value).withStreamKey(topic.getBytes());
// 刚追加记录的记录ID
RecordId recordId = connectionFactory.getConnection().xAdd(byteRecord);
return true;
}
}
消费者
定义消费者抽象接口，这个接口是为了后续手动订阅准备的。
/**
* 手动订阅消息总线消费者
*/
public interface DynamicSubscribeCustomer {
String consume(AdapterMessage message) throws IOException;
}
订阅者组内多播处理抽象类
public abstract class EventMessageBroker {
/**
* 消息多播
*
* 将集群间消息在自己服务内部广播，是的所有订阅消息都可以收到
*
* @param message
* @return
*/
public abstract void multicastEvent(AdapterMessage message) throws IllegalAccessException, InstantiationException, InvocationTargetException, IOException;
public void doMulticastEvent(AdapterMessage message, Map<TopicMatcher, Map<String, MessageAdapter>> subscribesPool, ApplicationContext applicationContext) throws InvocationTargetException, IllegalAccessException, InstantiationException, IOException {
// 查找相匹配的Adapter
List<MessageAdapter> adapterList = TopicUtil.getAllMatchedAdapter(message.getTopic(), subscribesPool);
for (MessageAdapter adapter : adapterList) {
if (adapter != null && adapter.isActive()) {
// 手动订阅消息
if (adapter.getCustomer() != null) {
adapter.getCustomer().consume(message);
} else {
// 取得对象
Object instance = applicationContext.getBean(adapter.getClazz());
if (instance != null) {
// 将消息转化为所需要的类型
if (adapter.getMethod().getParameterCount() > 0) {
Class<?> param = adapter.getMethod().getParameterTypes()[0];
adapter.getMethod().invoke(instance, JSON.parseObject(message.getPayload(), param));
} else {
adapter.getMethod().invoke(instance);
}
}
}
}
}
}
}
使用redis实现消费者
/**
* 消息总线-集群间消息redis实现
*
* 实现规则
* 根据不同的MQ机制实现消息的分组消费，将原始消息转化为AdapterMessage，然后使用继承于EventMessageBroker进行消息多播
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "stream.broker", name = "type", havingValue = "redis", matchIfMissing = true)
public class RedisMessageBroker extends EventMessageBroker implements StreamListener<String, MapRecord<String, String, String>> {
@Value("${stream.topic:/stream}")
String topic;
@Value("${stream.consumer.id}")
String consumerId;
@Value("${stream.timeout:10}")
Integer timeout;
@Value("${stream.consumer.group}")
String groupName = "stream.redis";
private final RedisTemplate<String, String> streamRedisTemplate;
private final ApplicationContext applicationContext;
private final EventBus eventBus;
public RedisMessageBroker(RedisTemplate<String, String> streamRedisTemplate, ApplicationContext applicationContext, EventBus eventBus) {
this.streamRedisTemplate = streamRedisTemplate;
this.applicationContext = applicationContext;
this.eventBus = eventBus;
}
@SneakyThrows
@Override
public void onMessage(MapRecord<String, String, String> message) {
log.info("消息内容-->{}", message.getValue());
StreamOperations<String, String, String> streamOperations = streamRedisTemplate.opsForStream();
// 服务内消息多播
AdapterMessage adapterMessage = new AdapterMessage();
adapterMessage.setTopic(message.getValue().get("topic"));
adapterMessage.setPayload(message.getValue().get("payload"));
try {
this.multicastEvent(adapterMessage);
} catch (Exception e) {
log.info("消息多播失败：" + e.getLocalizedMessage());
}
//消息应答
streamOperations.acknowledge(topic, groupName, message.getId());
}
@Bean
public StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> emailListenerContainerOptions() {
StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
return StreamMessageListenerContainer.StreamMessageListenerContainerOptions
.builder()
//block读取超时时间
.pollTimeout(Duration.ofSeconds(timeout))
//count 数量（一次只获取一条消息）
.batchSize(1)
//序列化规则
.serializer(stringRedisSerializer)
.build();
}
/**
* 开启监听器接收消息
*/
@Bean
public StreamMessageListenerContainer<String, MapRecord<String, String, String>> emailListenerContainer(RedisConnectionFactory factory,
StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> streamMessageListenerContainerOptions) {
StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = StreamMessageListenerContainer.create(factory,
streamMessageListenerContainerOptions);
//如果 流不存在 创建 stream 流
if (!streamRedisTemplate.hasKey(topic)) {
streamRedisTemplate.opsForStream().add(topic, Collections.singletonMap("", ""));
log.info("初始化集群间通信Topic{} success", topic);
}
//创建消费者组
try {
streamRedisTemplate.opsForStream().createGroup(topic, groupName);
} catch (Exception e) {
log.info("消费者组 {} 已存在", groupName);
}
//注册消费者 消费者名称，从哪条消息开始消费，消费者类
// > 表示没消费过的消息
// $ 表示最新的消息
listenerContainer.receive(
Consumer.from(groupName, consumerId),
StreamOffset.create(topic, ReadOffset.lastConsumed()),
this
);
listenerContainer.start();
return listenerContainer;
}
@Override
public void multicastEvent(AdapterMessage message) throws IllegalAccessException, InstantiationException, InvocationTargetException, IOException {
super.doMulticastEvent(message, eventBus.getSubscribesPool(), applicationContext);
}
}
配套@Subscribe注解
/**
* 自定义消息订阅
*
* @author liuhl
*/
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
@SuppressWarnings("unused")
public @interface Subscribe {
/**
* 订阅topic
* @return
*/
@AliasFor("value")
String topic() default "";
@AliasFor("topic")
String value() default "";
}
实现BeanPostProcessor监听所有的bean创建
@Component
@Slf4j
public class SpringMessageSubscribe implements BeanPostProcessor {
private final EventBus eventBus;
private final MessageAdapterCrater messageAdapterCrater = new DefaultMessageAdapterCrater();
public SpringMessageSubscribe(EventBus eventBus) {
this.eventBus = eventBus;
}
/**
* 为注解添加监听处理
*
* @param bean
* @param beanName
* @return
* @throws BeansException
*/
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
Class<?> type = ClassUtils.getUserClass(bean);
ReflectionUtils.doWithMethods(type, method -> {
// 取得所有订阅方法
AnnotationAttributes subscribes = AnnotatedElementUtils.getMergedAnnotationAttributes(method, Subscribe.class);
if (CollectionUtils.isEmpty(subscribes)) {
return;
}
// 生成订阅Adapter
MessageAdapter sub = messageAdapterCrater.createMessageAdapter(type, method, subscribes.getString("topic"));
// 注册订阅Adapter
eventBus.addAdapter(sub, subscribes.getString("topic"));
});
return bean;
}
}
EventBus对象
EventBus实体，主要负责三个功能，发布、订阅的入口，以及管理维护MessageAdapter
/**
* 消息总线
*/
@Component
public class EventBus {
private final MessagePublisher messagePublisher;
private Map<TopicMatcher, Map<String, MessageAdapter>> subscribesPool = new ConcurrentHashMap();
public EventBus(MessagePublisher messagePublisher) {
this.messagePublisher = messagePublisher;
}
/**
* 集群间消息发布
*
* @param message 消息体
* @return 发布结果
*/
public Boolean publish(AdapterMessage message) {
return messagePublisher.publish(message);
}
/**
* 将Adapter注册到EventBus
*
* @param adapter
* @param topic
*/
public void addAdapter(MessageAdapter adapter, String topic) {
if (subscribesPool.get(topic) != null) {
subscribesPool.get(topic).put(adapter.getId(), adapter);
} else {
Map<String, MessageAdapter> adapterMap = new HashMap<>();
adapterMap.put(adapter.getId(), adapter);
subscribesPool.put(new TopicMatcher(topic), adapterMap);
}
}
/**
* 将Adapter注册到EventBus
*
* @param adapter
* @param topic
*/
public void removeAdapter(MessageAdapter adapter, String topic) {
if (subscribesPool.get(topic) != null) {
subscribesPool.get(topic).remove(adapter.getId());
}
}
public Map<TopicMatcher, Map<String, MessageAdapter>> getSubscribesPool() {
return this.subscribesPool;
}
}
