实现原理

参考了使用Redis实现MQ

redis中已经实现了publish-subscribe，订阅者（Subscriber）可以订阅自己感兴趣的频道（Channel），发布者（Publisher）可以将消息发往指定的频道（Channel），正式通过这种方式，可以将消息的发送者和接收者解耦。另外，由于可以动态的Subscribe和Unsubscribe，也可以提高系统的灵活性和可扩展性。

打开redis客户端，使用SUBSCRIBE命令就可以订阅消息了，如：

SUBSCRIBE channel1 channel2

发布命令如下：

PUBLISH channel1 "Hello,world!"

这样在消息订阅的一方就可以接收到消息了，如下：

1) "message"
2) "channel1"
3) "Hello,world!"

在SpringBoot中使用

1
@Bean
2
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
3
  RedisMessageListenerContainer container = new RedisMessageListenerContainer();
4
  container.setConnectionFactory(connectionFactory);
5
  container.addMessageListener(listenerAdapter, new PatternTopic("channel1"));
6
  return container;
7
}
8

9
@Bean //一定要将消息适配器注册到spring容器
10
MessageListenerAdapter listenerAdapter(Receiver receiver/*自定义的类的对象*/) {
11
  return new MessageListenerAdapter(receiver, "receiveMessage"/*反射调用这个对象的这个方法*/);
12
}

ok,也就是只要提供一个RedisMessageListenerContainer和一堆MessageListenerAdapter给spring即可而这些完全可以通过给方法加注解+反射直接创建出来

理论存在,实践开始！

首先需要拿到所有带有 @RedisMessageListener 注解的方法在苦苦搜寻了一番后，在stackoverflow上找到了org.reflections.Reflections这个类（maven依赖地址），使用下面的方式就ok了

1
Set<Method> allMethods = new Reflections("扫描包名", new MethodAnnotationsScanner()).getMethodsAnnotatedWith(RedisMessageListener.class);

那么应该如何直接获取当前项目的basePackage呢？又苦苦搜寻了一番后，终于在springboot的issues中找到了AutoConfigurationPackages.get() 看一下所需参数，BeanFactory！ok，从applicationContext中掏出来传给他打印输出返回值 [com.coderxi]，完美！拿到第一个即可！

1
String baskPackage = AutoConfigurationPackages.get(applicationContext.getBeanFactory()).get(0);

拿到了方法，也就能够拿到方法所在的类，再从容器中实例，就满足了构造MessageListenerAdapter的参数但是，这个MessageListenerAdapter必须注册到容器
那么应该如何使用Java代码手动注册对象到Spring容器呢？第三次苦苦搜寻了一番后，又通过尝试，找到了最优雅最简洁的方式使用 AnnotationConfigApplicationContext 的 registerBean 注册进去，然后get回来就ok 实例的名字不能重复，我就直接采用了 方法名+类名+"MessageListenerAdapter" 作为实例名
最后，按照上面的 RedisMessageListenerContainer 的配置方式将所有 MessageListenerAdapter 实例和对应的 channel（也就是topic）交给他即可

成果

1
import org.reflections.Reflections;
2
import org.reflections.scanners.MethodAnnotationsScanner;
3
import org.springframework.boot.autoconfigure.AutoConfigurationPackages;
4
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
5
import org.springframework.context.annotation.Bean;
6
import org.springframework.context.annotation.Configuration;
7
import org.springframework.data.redis.connection.RedisConnectionFactory;
8
import org.springframework.data.redis.listener.PatternTopic;
9
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
10
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
11

12
import java.lang.reflect.Method;
13
import java.util.LinkedHashMap;
14
import java.util.Map;
15
import java.util.Set;
16

17
@Configuration
18
public class RedisMessageListenerAdapterConfiguration {
19

20
  @Bean
21
  RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, Map<MessageListenerAdapter, String[]> redisMessageListenerAdapterAndTopics) {
22
    RedisMessageListenerContainer container = new RedisMessageListenerContainer();
23
    container.setConnectionFactory(connectionFactory);
24
    for (Map.Entry<MessageListenerAdapter, String[]> entry : redisMessageListenerAdapterAndTopics.entrySet()) {
25
      for (String topic : entry.getValue()) {
26
        container.addMessageListener(entry.getKey(), new PatternTopic(topic));
27
      }
28
    }
29
    return container;
30
  }
31

32
  @Bean
33
  Map<MessageListenerAdapter, String[]> redisMessageListenerAdapterAndTopics(AnnotationConfigApplicationContext applicationContext) {
34
    Map<MessageListenerAdapter, String[]> result = new LinkedHashMap<>();
35
    String baskPackage = AutoConfigurationPackages.get(applicationContext.getBeanFactory()).get(0);
36
    Set<Method> allMethods = new Reflections(baskPackage, new MethodAnnotationsScanner()).getMethodsAnnotatedWith(RedisMessageListener.class);
37
    for (Method method : allMethods) {
38
      RedisMessageListener annotation = method.getAnnotation(RedisMessageListener.class);
39
      String newMessageListenerAdapterName = method.getName() + method.getDeclaringClass().getSimpleName() + "MessageListenerAdapter";
40
      applicationContext.registerBean(newMessageListenerAdapterName, MessageListenerAdapter.class, applicationContext.getBean(method.getDeclaringClass()), method.getName());
41
      result.put(applicationContext.getBean(newMessageListenerAdapterName,MessageListenerAdapter.class),annotation.channels());
42
    }
43
    return result;
44
  }
45

46
}

1
@Retention(RetentionPolicy.RUNTIME)
2
@Target(ElementType.METHOD)
3
public @interface RedisMessageListener {
4
  String[] channels();
5
}

1
import org.junit.jupiter.api.Test;
2
import org.springframework.beans.factory.annotation.Autowired;
3
import org.springframework.boot.test.context.SpringBootTest;
4
import org.springframework.data.redis.core.StringRedisTemplate;
5

6
@SpringBootTest
7
public class Sender {
8

9
  @Autowired
10
  StringRedisTemplate redisTemplate;
11

12
  @Test
13
  void send(){
14
    redisTemplate.convertAndSend("channel1","Hello,world!");
15
  }
16

17
}

1
import com.coderxi.redismq.RedisMessageListener;
2
import org.springframework.stereotype.Component;
3

4
@Component
5
public class Receiver {
6

7
  @RedisMessageListener(channels = "channel1")
8
  public void receive1(String message) {
9
    System.out.println("receive1: " + message);
10
  }
11

12
  @RedisMessageListener(channels = {"channel1","channel2"})
13
  public void receive2(String message) {
14
    System.out.println("receive2: " + message);
15
  }
16

17
}

向channel1频道发送消息，我们可以看到两个都响应了

receive2: Hello,world!
receive1: Hello,world!

向channel2频道发送消息，我们可以看到只有receive2响应了

receive2: Hello,world!

Redis+SpringBoot+自定义注解实现消息发布订阅

实现原理

在SpringBoot中使用

理论存在,实践开始！

成果

评论

Redis+SpringBoot+自定义注解实现消息发布订阅

实现原理

Redis中的publish-subscribe

在SpringBoot中使用

理论存在,实践开始！

成果

评论