-1

Axon 消息接收但未调用事件处理程序。

我正在尝试使用两个不同的队列在两侧实现事件源。我的第一个队列是测试，第二个是测试演示

我有两个单独的应用程序在同一台服务器上运行。

  1. 用户管理
  2. 钱包管理

我已经实现了从用户管理到钱包管理的事件溯源。它工作正常。

现在我正在尝试对 UserManagement 实施钱包管理，这意味着当我将从钱包管理 (Producer) 和 (Consume) 用户管理应用程序发布事件时。因此接收到事件但未调用事件处理程序。

以下是我的应用程序代码。请帮我弄清楚我会错过什么。

我的 Axon 配置类

package com.peaas.ngapblueprintdemo.config;

import org.axonframework.amqp.eventhandling.DefaultAMQPMessageConverter;
import org.axonframework.amqp.eventhandling.spring.SpringAMQPMessageSource;
import org.axonframework.serialization.Serializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.ExchangeBuilder;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.rabbitmq.client.Channel;

@Configuration
public class AxonConfiguration {

    private final static Logger logger = LoggerFactory.getLogger(AxonConfiguration.class);

    @Value("${axon.amqp.exchange}")
    private String exchange;

    @Bean
    public Exchange exchange() {
        logger.info(exchange + " AMQP Exchange Registering ");
        return ExchangeBuilder.fanoutExchange(exchange).build();
    }

    @Bean
    public Queue queue() {
        return QueueBuilder.durable(exchange).build();
    }

    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue()).to(exchange()).with("*").noargs();
    }

    @Autowired
    public void configure(AmqpAdmin amqpAdmin) {
        amqpAdmin.declareExchange(exchange());
        amqpAdmin.declareQueue(queue());
        amqpAdmin.declareBinding(binding());
    }   

    @Bean
    public SpringAMQPMessageSource testdemo(Serializer serializer) {
        System.out.println("--- On Message Call ---");
        return new SpringAMQPMessageSource(new DefaultAMQPMessageConverter(serializer)) {

            @RabbitListener(queues = "testdemo")

            @Override
            public void onMessage(Message message, Channel channel) throws Exception {
                System.out.println(message.getMessageProperties());
                System.out.println("channel == "+channel);
                super.onMessage(message, channel);
            }
        };
    }
}

WalletCreatedEvent 类

package com.peaas.ngapblueprintdemo.events;

public class WalletCreatedEvent {
    private Long id;
    private String walletId;
    private Double amount;
    private Long userId;

    public WalletCreatedEvent(Long id, String walletId, Double amount, Long userId) {       
        super();
        System.out.println("--- call ---");
        this.id = id;
        this.walletId = walletId;
        this.amount = amount;
        this.userId = userId;
    }

    public Long getId() {
        return id;
    }

    public void setId(Long id) {
        this.id = id;
    }

    public Double getAmount() {
        return amount;
    }

    public void setAmount(Double amount) {
        this.amount = amount;
    }

    public Long getUserId() {
        return userId;
    }

    public void setUserId(Long userId) {
        this.userId = userId;
    }

    public String getWalletId() {
        return walletId;
    }

    public void setWalletId(String walletId) {
        this.walletId = walletId;
    }

    @Override
    public String toString() {
        return "WalletCreatedEvent [id=" + id + ", walletId=" + walletId + ", amount=" + amount + ", userId=" + userId
                + "]";
    }

}

事件处理程序类

package com.peaas.ngapblueprintdemo.eventHandlers;

import org.axonframework.eventhandling.EventHandler;
import org.springframework.stereotype.Component;

import com.peaas.ngapblueprintdemo.events.WalletCreatedEvent;

@Component
public class UserEventHandler {

    @EventHandler
    public void onCreateWalletEvent(WalletCreatedEvent event) {
        System.out.println("--- Wallet Created Successfully ---");
        System.out.println(event);
    }   
}

以下是我的 application.yml 文件属性

axon:
    amqp:
        exchange: test
    eventhandling:
        processors:
            amqpEvents:
                source: testdemo

以下是我收到显示事件的日志数据。

MessageProperties [headers={axon-message-id=fa60968c-6905-46b5-8afe-6da853a4c51a, axon-message-aggregate-seq=0, axon-metadata-correlationId=589ef284-176f-49b8-aae0-0ad1588fa735, axon-message-aggregate-type=WalletAggregate, axon-message-revision=null, axon-message-timestamp=2018-08-06T11:09:26.345Z, axon-message-type=com.peaas.ngapblueprintdemo.events.WalletCreatedEvent, axon-metadata-traceId=589ef284-176f-49b8-aae0-0ad1588fa735, axon-message-aggregate-id=9524f7df-44fb-477f-83b8-d176583a126e}, contentLength=0, receivedDeliveryMode=PERSISTENT, redelivered=false, receivedExchange=testdemo, receivedRoutingKey=com.peaas.ngapblueprintdemo.events, deliveryTag=1, consumerTag=amq.ctag-fGm3jQcP_JIoTGf4ZMhAIg, consumerQueue=testdemo]
channel == Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,1), conn: Proxy@3dcd657d Shared Rabbit Connection: SimpleConnection@19b12fd2 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 52963]
4

1 回答 1

0

您拥有大部分正确的配置，但您忘记将您的配置绑定SpringAMQPMessageSource到放置事件处理组件的事件处理器。

有关如何达到此目的的正确示例，请参阅参考指南。

以下是该参考指南中用于将消息源配置为事件处理器的直接片段：

@Autowired
public void configure(EventHandlingConfiguration ehConfig, SpringAmqpMessageSource myMessageSource) {
ehConfig.registerSubscribingEventProcessor("myProcessor", c -> myMessageSource);
}

编辑

我想我知道你错过了哪一部分。您确实正确地将队列作为可订阅消息源连接到事件处理器。这是从你那里得到application.yml的，它将testdemo消息源与amqpEvents事件处理器联系起来。因此，我对我之前关于这部分的假设感到抱歉。

您没有在UserEventHandler, 中收到事件的原因是该事件处理程序未绑定到amqpEvents事件处理器。要解决这个问题，您应该将@ProcessingGroup("amqpEvents")注释添加到UserEventHandler组件中。

于 2018-08-06T11:54:27.910 回答