如何在 CQRS 中访问命令总线

如何解决如何在 CQRS 中访问命令总线

完整代码位于：https://github.com/Naresh-Chaurasia/API-MicroServices-Kafka/tree/master/Microservices-CQRS-SAGA-Kafka

我有以下用例图：

ProductsCommandController

@RestController
@RequestMapping("/products")
public class ProductsCommandController {

    private final Environment env;
    private final CommandGateway commandGateway;

    @Autowired
    public ProductsCommandController(Environment env,CommandGateway commandGateway) {
        this.env = env;
        this.commandGateway = commandGateway;
    }

    @PostMapping
    public String createProducts(@RequestBody CreateProductRestModel createProductRestModel) throws InterruptedException {

        CreateProductCommand createProductCommand = CreateProductCommand.builder()
                .price(createProductRestModel.getPrice())
                .quantity(createProductRestModel.getQuantity())
                .title(createProductRestModel.getTitle())
                .productId(UUID.randomUUID().toString()).build();

        String returnValue = commandGateway.sendAndWait(createProductCommand);

        return returnValue;

    }

}

创建产品命令

@Builder
@Data
public class CreateProductCommand {
    
    @TargetAggregateIdentifier
    private final String productId;
    private final String title;
    private final BigDecimal price;
    private final Integer quantity;
    
}

ProductEventsHandler

@Component
public class ProductEventsHandler {

    private final ProductsRepository productsRepository;
    private static final Logger LOGGER = LoggerFactory.getLogger(ProductEventsHandler.class);

    public ProductEventsHandler(ProductsRepository productsRepository) {
        this.productsRepository = productsRepository;
    }

    @EventHandler
    public void on(ProductCreatedEvent event) {

        ProductEntity productEntity = new ProductEntity();
        BeanUtils.copyProperties(event,productEntity);

        try {
            productsRepository.save(productEntity);
        } catch (IllegalArgumentException ex) {
            ex.printstacktrace();
        }

    }
}

产品存储库

public interface ProductsRepository extends JpaRepository<ProductEntity,String> {
    
    ProductEntity findByProductId(String productId);
    ProductEntity findByProductIdOrTitle(String productId,String title);

}

在研究 CQRS 时，我遇到了以下声明：

现在我们已经创建了命令对象，我们准备将此命令发送到命令网关，我们将命令网关对象注入到我们的控制器类中。命令网关是一个对象，用于将我们创建的这个命令对象发送到命令总线。因此，将命令网关视为发送命令的 API，命令总线是一种接收此命令并将此命令路由到命令处理程序的机制。

我的问题如下：

创建了什么样的命令总线。我们可以用队列/主题/kafka 等其他设备替换这条总线吗？如果是，我们该怎么做。

还有，可以看到公交车的内容吗。

请指导。

提前致谢。

解决方法

Kafka 等作为该架构中的命令总线非常有用。在 Kafka 的情况下，一个主题（或者更准确地说是一个主题的一个分区，因为排序只在一个分区内得到保证）通常是一个命令总线。

通常，命令总线（因为它似乎在此上下文中使用）可能不允许查看其内容（例如，命令可以通过 HTTP 端点发送；除非该端点的处理程序也碰巧将命令保存在某处（一个命令源的示例），您将无法（缺少数据包捕获等内容）查看总线的内容）；在设计/选择/实施一个命令总线时，命令总线是否应该自省和/或持久绝对是一个考虑因素。