Vor kurzem habe ich ein altes System renoviert und bin auf ein Szenario gestoßen, in dem im vorherigen Prozess verschiedene Konfigurationen auf der Sende- und Konsumseite erforderlich waren. Dann fiel mir plötzlich die Form von @Reference ein Annotation in Dubbo, damit das Aufrufen von MQ so bequem und einfach ist wie das Aufrufen der synchronen Schnittstelle. Ich habe also die relevanten Informationen überprüft und den Quellcode von Dubbo gelesen 🎜> Zusammenfassend lässt sich sagen, dass das zu erreichende Ziel wie bei Dubbo ist: Die Verbraucherseite stellt die Schnittstelle bereit (Sie können sogar die durch den Dubbo-Dienst definierte Schnittstelle wiederverwenden, sodass das Schreiben eines Dubbo-Dienstes synchron oder MQ asynchron sein kann) und das Senden Seite injiziert Objektaufrufe über benutzerdefinierte Annotationen. Die Methode wird intern vom Framework verarbeitet und dann in eine asynchrone MQ-Form konvertiert und an den Verbraucher gesendet.
Der Server verfügt beispielsweise über eine Schnittstelle:
und hat eine Implementierung:public interface MqDemoService { void dealById(Long id); }Darunter:
@Slf4j @Component("mqDemoServiceImpl") @Service(version = "1.0.0") public class MqDemoServiceImpl implements MqDemoService { @Override public void dealById(Long id) { log.info("执行findById方法"); } }Interessierte Studierende können
@Slf4j是lombok注解 @Service是dubbo服务端注解
selbst überprüfen und dann den Versand beenden
hat benutzerdefinierte Annotationen:
also beim Aufruf im Controller:@Target(ElementType.FIELD) @Retention(RetentionPolicy.RUNTIME) @Documented @Inherited public @interface AsyncInvoker { }Achten Sie auf das Attribut mqDemoService, das von @AsyncInvoker im Controller annotiert wird, wenn das Objekt durchgeleitet wird Diese Annotation ruft die Methode auf, sie wird über mq gesendet und zu einem asynchronen Aufruf.
@Slf4j @Controller public class MqDemoController { @AsyncInvoker private MqDemoService mqDemoService; @RequestMapping(value = "/deal", method = RequestMethod.POST) public void deal() { mqDemoService.dealById(1L); } }
Okay, das zu erreichende Ziel ist ganz klar die folgenden Aspekte:
1. So bestimmen Sie das Format der gesendeten Nachricht, damit der Empfänger die aufzurufende Methode bestimmen kann1,如何确定发送消息的格式,使消费端可以确定调用的方法 2,发送端中如何为注解@AsyncInvoker注释的对象注入实例 3,接收端中如何在接收到消息后调用对应接口的实现方法 4,多个消费服务如何区分mq队列.
@Data public class MqMethodMeta { //调用的接口名称(包括包名,用于反射) private String interfaceName; //调用的方法名 private String methodName; //调用的方法的参数 private Object[] args; //调用的方法的参数类型 private String[] paramTypeNames; }
Was soll Spring also noch wissen lassen, dass das mit @AsyncInvoker annotierte Objekt mit einem dynamischen Proxy injiziert werden muss? 🎜>
Die Antwort ist die BeanPostProcessor-Schnittstelle von Spring. Diese Schnittstelle ermöglicht es Spring, benutzerdefinierte Logik vor und nach der Verarbeitung von Objekten einzufügen, die hier nicht im Detail erläutert werden 🎜>@Slf4j @Component public class AsyncInvokerBeanProcessor implements BeanPostProcessor { //缓存生成的动态代理对象,用于多个Controller注入同一类型对象时使用. private final ConcurrentMap<String, Object> proxyMap = new ConcurrentHashMap<>(); //注入spring amqp处理mq的对象 @Autowired private RabbitTemplate rabbitTemplate; //BeanPostProcessor接口方法,在spring创建每个实例前插入的用户自定义逻辑.这里我们需要的是在每个Controller对象创建的时候为其中的@AsyncInvoker注解对象注入动态代理. @Override public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { //获取该实例中的有@AsyncInvoker注解的field Field[] fields = bean.getClass().getDeclaredFields(); for (Field field : fields) { try { if (!field.isAccessible()) { field.setAccessible(true); } AsyncInvoker asyncInvoker = field.getAnnotation(AsyncInvoker.class); if (asyncInvoker != null) { //创建代理对象,赋值给该feild Object value = createProxy(field.getType()); if (value != null) { field.set(bean, value); } } } catch (Throwable e) { log.error("Failed to init remote mq service at filed " + field.getName() + " in class " + bean.getClass().getName() + ", cause: " + e.getMessage(), e); } } return bean; } @Override public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { return bean; } private Object createProxy(Class clz) { String interfaceName; if (clz.isInterface()) { interfaceName = clz.getName(); } else { throw new IllegalStateException("The @MqInvoker property type " + clz.getName() + " is not a interface."); } Object proxy = proxyMap.get(interfaceName); if (proxy == null) { Object newProxy = Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{clz}, new InvocationHandler() { @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { log.debug("执行动态代理! method:{} ,args: {}", method, args); if (method.getParameters().length != 1 || !method.getParameters()[0].getType().equals(Long.class)) { throw new IllegalAccessException("MQ Service 目前仅支持单参数Long类型方法"); } //动态代理中创建mq传输对象并发送. MqMethodMeta mqMethodMeta = new MqMethodMeta(); mqMethodMeta.setInterfaceName(clz.getName()); mqMethodMeta.setMethodName(method.getName()); mqMethodMeta.setArgs(args); String[] paramTypeNames = new String[args.length]; for (int i = 0; i < args.length; i++) { paramTypeNames[i] = args[i].getClass().getName(); } mqMethodMeta.setParamTypeNames(paramTypeNames); RabbitAdmin admin = new RabbitAdmin(rabbitTemplate.getConnectionFactory()); Exchange exchange = new TopicExchange("exchange.demo.web.adaptor"); admin.declareExchange(exchange); //关注此处clz.getName(),用于处理问题4 rabbitTemplate.convertAndSend("exchange.demo.web.adaptor", clz.getName(), mqMethodMeta); return null; } }); proxyMap.putIfAbsent(interfaceName, newProxy); proxy = proxyMap.get(interfaceName); } return proxy; } }4, mehr MQ-Warteschlangen unterscheiden? Hier wird der Thementyp „Kaninchen“ verwendet.
@Slf4j public class AsyncMethodListener implements ApplicationContextAware { private ApplicationContext applicationContext; @RabbitListener(bindings = @QueueBinding( value = @Queue(value = "${demo.mq.method.queue}", durable = "true"), exchange = @Exchange(value = "exchange.demo.web.adaptor", type = ExchangeTypes.TOPIC, durable = "true"), key = "${demo.mq.method.routekey}" )) public void messageHandle(@Payload MqMethodMeta message) { try { log.info("收到message: {}", message); Class clz = Class.forName(message.getInterfaceName()); String methodName = message.getMethodName(); Object[] args = message.getArgs(); Class[] paramTypes = new Class[message.getParamTypeNames().length]; for (int i = 0; i < message.getParamTypeNames().length; i++) { paramTypes[i] = Class.forName(message.getParamTypeNames()[i]); } //由于使用Object[]数组传送参数,所以Jackson2JsonMessageConverter会将id转换为Integer,反射调用时会报错,此处强转一下 for (int i = 0; i < args.length; i++) { Class c = paramTypes[i]; if (args[i] instanceof Integer && c.equals(Long.class)) { args[i] = ((Integer) args[i]).longValue(); } } //拿到spring管理的对应接口的实现 Object invoker = applicationContext.getBean(clz); Method method = clz.getMethod(methodName, paramTypes); method.invoke(invoker, args); } catch (Exception e) { e.printStackTrace(); } } @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = applicationContext; } }Konfigurieren Sie zunächst die Warteschlange und den Routenschlüssel im verbraucherseitigen Listener:
Beachten Sie, dass
@RabbitListener(bindings = @QueueBinding( value = @Queue(value = "${demo.mq.method.queue}", durable = "true"), exchange = @Exchange(value = "exchange.demo.web.adaptor", type = ExchangeTypes.TOPIC, durable = "true"), key = "${demo.mq.method.routekey}" ))hier wird aus der Konfigurationsdatei gelesen: Zum Beispiel ist die Konfiguration in System 1 wie folgt:
${demo.mq.method.queue} ${demo.mq.method.routekey}In System 2 ist die Konfiguration wie folgt :
demo.mq.method.queue=com.demo.service.project1.# demo.mq.method.routekey=com.demo.service.project1.#Sehen Sie sich den Code im Absender an:
demo.mq.method.queue=com.demo.service.project2.# demo.mq.method.routekey=com.demo.service.project2.#Der darin enthaltene clz.getName() Da unser System über eine gute Subunternehmerstrategie verfügt, ist der clz.getName(. ) von System 1 muss mit com.demo.service.project1 beginnen. Es wird definitiv an den Listener in project1 gesendet. Der Wert von clz.getName() ist beispielsweise com.demo.service.project1.MqDemoService (. #" stimmt mit mehreren Bezeichnern am Ende überein, was eine Funktion des Thementypaustauschs in RabbitMQ ist). An diesem Punkt wurde das Ziel erreicht, das Sie zu Beginn erreichen wollten. In der Zukunft, wenn Sie Sie müssen mq verwenden, um asynchrone Aufrufe durchzuführen. Es kann wie eine synchrone Methode verwendet werden.
//关注此处clz.getName(),用于处理问题4 rabbitTemplate.convertAndSend("exchange.demo.web.adaptor", clz.getName(), mqMethodMeta);
Die Verwendung von mq im Frühjahr wird hier nicht im Detail aufgeführt. Sie können sich auf das Dokument beziehen:
Ein Satz Democode wird später zur Aufzeichnung und Referenz bereitgestellt.Zusammenfassung
http://docs.spring.io/spring-amqp/docs/1.7.3.RELEASE/reference/htmlsingle/
Bei dieser Methode gibt es noch einige Probleme:
Das obige ist der detaillierte Inhalt vonVerwenden Sie Rabbit MQ, um Dubbo zu simulieren und asynchrone MQ-Aufrufe durchzuführen. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!