Verwenden Sie Rabbit MQ, um Dubbo zu simulieren und asynchrone MQ-Aufrufe durchzuführen

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 kann

1,如何确定发送消息的格式,使消费端可以确定调用的方法
2,发送端中如何为注解@AsyncInvoker注释的对象注入实例
3,接收端中如何在接收到消息后调用对应接口的实现方法
4,多个消费服务如何区分mq队列.

Hier definiere ich einfach ein Übertragungsobjekt gemäß den für Java-Reflection-Aufrufe erforderlichen Parametern:

2. So injizieren Sie Instanzen von mit @AsyncInvoker versehenen Objekten in den Absender

@Data
public class MqMethodMeta {
  //调用的接口名称(包括包名,用于反射)
  private String interfaceName;
  //调用的方法名
  private String methodName;
  //调用的方法的参数
  private Object[] args;
  //调用的方法的参数类型
  private String[] paramTypeNames;
}

In diesem Szenario führt der Absender nur die Schnittstelle des Verbrauchers ein, nicht die Implementierung Wie fügt @AsyncInvoker das Objekt ein?

Die Antwort ist ein dynamischer Proxy.

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 🎜>

Dann kommt die Idee heraus, der Code lautet wie folgt:

3. So rufen Sie das Empfangsende nach dem Empfang der Nachricht auf. Die Implementierungsmethode der entsprechenden Schnittstelle

Für den Empfänger ist es sehr einfach, die entsprechende Schnittstelle aufzurufen. Er muss lediglich das MqMethodMeta-Objekt abrufen und einen Reflexionsaufruf durchführen:

@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 erste Problem muss das entsprechende Unternehmen verwenden. Lassen Sie uns später darüber nachdenken, oder Sie können es besprechen, wenn Sie einige Ideen haben.

Die zweite Frage besteht hauptsächlich darin, zu überlegen, ob der Verbraucher den Parametertyp ändert Andere Situationen, nach der Neuveröffentlichung, für mögliche Kompatibilität alter Nachrichten, die im MQ verbleiben. Im Moment gibt es dafür wirklich keine gute Idee, und es dient nur dem Brainstorming

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!