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);
}@Slf4j
@Component("mqDemoServiceImpl")
@Service(version = "1.0.0")
public class MqDemoServiceImpl implements MqDemoService {
@Override
public void dealById(Long id) {
log.info("执行findById方法");
}
}@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 {
}@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;
}
}@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;
}
}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}"
))${demo.mq.method.queue}
${demo.mq.method.routekey}demo.mq.method.queue=com.demo.service.project1.# demo.mq.method.routekey=com.demo.service.project1.#
demo.mq.method.queue=com.demo.service.project2.# demo.mq.method.routekey=com.demo.service.project2.#
//关注此处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!
Was ist der Unterschied zwischen Dubbo und Zookeeper?
Was sind die Unterschiede zwischen Springcloud und Dubbo?
Was ist das Prinzip und der Mechanismus von Dubbo?
So löschen Sie die letzte leere Seite in Word
msvcp140.dll
Der Unterschied zwischen PD-Schnellladen und allgemeinem Schnellladen
So lösen Sie das Problem, dass in Win10 plötzlich nicht mehr alle Ordner geöffnet werden können
Warum hat der Computer einen blauen Bildschirm?
