Axon入门系列(五)：第一个正式Axon例子

原创
Java框架CQRSAxon
CQRS, DDD, axon, event sourcing20857 人阅读

前面对Axon的基本概念和基本操作做了简介，从本章开始，我们将一步步使用AxonFramework完成一个真正CQRS&EventSourcing的例子。

设计

回顾一下使用AxonFramework应用的架构

Command端Repository和Query端的Database是解耦的，完全可以使用不同的持久化技术，我们来尝试用MongoDB做Command端的Repository，而MySQL做Query的数据库。

例子描述

我们尝试完成一个简单的case：后台人员创建商品，用户选定若干商品后下单购买。
商品定义：Product(id, name, stock, price)
商品创建流程：
CreateProductCommand -> new ProductAggregate instance -> ProductCreatedEvent

订单定义： Order(id, username, payment, products)
订单创建流程：
CreateOrderCommand -> new OrderAggregate instance -> OrderCreatedEvent
创建商品时，我们只接收商品ID，去查询商品的具体信息，这样来学习如何在handler内去查询Aggregate。

Command端实现

Command端实现与前面几篇文章基本一致，需要定义Aggregate、Command，然后提供配置即可。

Aggregate

ProductAggregate

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
@Aggregate
public class ProductAggregate {

    private static final Logger LOGGER = getLogger(ProductAggregate.class);

    @AggregateIdentifier
    private String id;
    private String name;
    private int stock;
    private long price;

    public ProductAggregate() {
    }

    @CommandHandler
    public ProductAggregate(CreateProductCommand command) {
        apply(new ProductCreatedEvent(command.getId(),command.getName(),command.getPrice(),command.getStock()));
    }

    @EventHandler
    public void on(ProductCreatedEvent event){
        this.id = event.getId();
        this.name = event.getName();
        this.price = event.getPrice();
        this.stock = event.getStock();
        LOGGER.debug("Product [{}] {} {}x{} is created.", id,name,price,stock);
    }

    // getter and setter
    ......
}

OrderAggregate

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@Aggregate
public class OrderAggregate {

    @AggregateIdentifier
    private OrderId id;
    private String username;
    private double payment;

    @AggregateMember
    private Map<String, OrderProduct> products;

    public OrderAggregate(){}

    public OrderAggregate(OrderId id, String username, Map<String, OrderProduct> products) {
        apply(new OrderCreatedEvent(id, username, products));
    }

    public OrderId getId() {
        return id;
    }

    public String getUsername() {
        return username;
    }

    public Map<String, OrderProduct> getProducts() {
        return products;
    }

    @EventHandler
    public void on(OrderCreatedEvent event){
        this.id = event.getOrderId();
        this.username = event.getUsername();
        this.products = event.getProducts();
        computePrice();
    }

    private void computePrice() {
        products.forEach((id, product) -> {
            payment += product.getPrice() * product.getAmount();
        });
    }

    /**
     * Divided 100 here because of the transformation of accuracy
     *
     * @return
     */
    public double getPayment() {
        return payment/100;
    }

    public void addProduct(OrderProduct product){
        this.products.put(product.getId(), product);
        payment += product.getPrice() * product.getAmount();
    }

    public void removeProduct(String productId){
        OrderProduct product = this.products.remove(productId);
        payment = payment - product.getPrice() * product.getAmount();
    }
}

这里，我并没有像ProductAggregate一样，把CreateOrderCommand放到OrderAggregate的构造器中去处理，原因是在创建订单时，由于需要知道商品的单价，所以要根据商品id查询商品信息，因为涉及到了其他Aggregate操作，特地单独创建一个OrderHandler来处理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
@Component
public class OrderHandler {

    private static final Logger LOGGER = getLogger(OrderHandler.class);

    @Autowired
    private Repository<OrderAggregate> repository;

    @Autowired
    private Repository<ProductAggregate> productRepository;

    @Autowired
    private EventBus eventBus;

    @CommandHandler
    public void handle(CreateOrderCommand command) throws Exception {
        Map<String, OrderProduct> products = new HashMap<>();
        command.getProducts().forEach((productId,number)->{
            LOGGER.debug("Loading product information with productId: {}",productId);
            Aggregate<ProductAggregate> aggregate = productRepository.load(productId);
            products.put(productId,
                    new OrderProduct(productId,
                            aggregate.invoke(productAggregate -> productAggregate.getName()),
                            aggregate.invoke(productAggregate -> productAggregate.getPrice()),
                            number));
        });
        repository.newInstance(() -> new OrderAggregate(command.getOrderId(), command.getUsername(), products));
    }
}

如果查看org.axonframework.commandhandling.model.Repository<T>接口的定义，会发现里面只有三个方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
public interface Repository<T> {

    /**
     * Load the aggregate with the given unique identifier. No version checks are done when loading an aggregate,
     * meaning that concurrent access will not be checked for.
     *
     * @param aggregateIdentifier The identifier of the aggregate to load
     * @return The aggregate root with the given identifier.
     * @throws AggregateNotFoundException if aggregate with given id cannot be found
     */
    Aggregate<T> load(String aggregateIdentifier);

    /**
     * Load the aggregate with the given unique identifier.
     *
     * @param aggregateIdentifier The identifier of the aggregate to load
     * @param expectedVersion     The expected version of the loaded aggregate
     * @return The aggregate root with the given identifier.
     * @throws AggregateNotFoundException if aggregate with given id cannot be found
     */
    Aggregate<T> load(String aggregateIdentifier, Long expectedVersion);

    /**
     * Creates a new managed instance for the aggregate, using the given {@code factoryMethod}
     * to instantiate the aggregate's root.
     *
     * @param factoryMethod The method to create the aggregate's root instance
     * @return an Aggregate instance describing the aggregate's state
     * @throws Exception when the factoryMethod throws an exception
     */
    Aggregate<T> newInstance(Callable<T> factoryMethod) throws Exception;
}

有人会疑惑了，为什么没有Delete和Update？
先说update，这个Repository其实是对Aggregate的操作，EventSourcing中对Aggregate所有的变化都是通过Event来实现的，所以在调用apply(EventMessage)时，Event就已经被持久化了，EventHandler在处理该Event时，就已经实现了对Aggregate的update。
而Delete没有，很简单，EventSourcing脱胎于现实概念，你见过现实生活中把一个事物真正“delete”掉吗？估计得使用高能量子炮把东西轰成原子吧。
所以，只会有一个把这个Aggregate标为失效的标志，Axon中，在Aggregate内部可以直接调用markDeleted()来表示这个Aggregate被“delete”掉了，其实只是不能被load出来罢了。
由于Repository默认返回的是同一类型Aggregate，所以我们取属性就没那么简单了，只能通过invoke来调用get方法。是不是觉得很麻烦？因为其实CQRS压根不推荐直接从Repository直接query Aggregate来查询，而是调用Query端。

Command

command的实现因为都是POJO我就不贴代码了，可以直接看源码。
这里写一下基于SpringWeb的Controller类（引入spring-boot-starter-web包），以创建Product为例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@RestController
@RequestMapping("/product")
public class ProductController {

    private static final Logger LOGGER = getLogger(ProductController.class);

    @Autowired
    private CommandGateway commandGateway;

    @RequestMapping(value = "/{id}", method = RequestMethod.POST)
    public void create(@PathVariable(value = "id") String id,
                       @RequestParam(value = "name", required = true) String name,
                       @RequestParam(value = "price", required = true) long price,
                       @RequestParam(value = "stock",required = true) int stock,
                       HttpServletResponse response) {

        LOGGER.debug("Adding Product [{}] '{}' {}x{}", id, name, price, stock);

        try {
            // multiply 100 on the price to avoid float number
            CreateProductCommand command = new CreateProductCommand(id,name,price*100,stock);
            commandGateway.sendAndWait(command);
            response.setStatus(HttpServletResponse.SC_CREATED);// Set up the 201 CREATED response
            return;
        } catch (CommandExecutionException cex) {
            LOGGER.warn("Add Command FAILED with Message: {}", cex.getMessage());
            response.setStatus(HttpServletResponse.SC_BAD_REQUEST);
            if (null != cex.getCause()) {
                LOGGER.warn("Caused by: {} {}", cex.getCause().getClass().getName(), cex.getCause().getMessage());
                if (cex.getCause() instanceof ConcurrencyException) {
                    LOGGER.warn("A duplicate product with the same ID [{}] already exists.", id);
                    response.setStatus(HttpServletResponse.SC_CONFLICT);
                }
            }
        }
    }
}

CommandGateway提供了四种发送Comman的方法：

  • send(command, CommandCallback) 发送command，根据执行结果调用CommandCallback中的onSuccess或onFailure方法
  • sendAndWait(command) 发送完command，等待执行完成并返回结果
  • sendAndWait(command, timeout, TimeUnit) 这个好理解，比上面多了一个超时
  • send(command) 该方法返回一个CompletableFuture，不用等待command的执行，立刻返回。结果通过future获取。

Repository

由于我们要使用axon-mongo，而非默认的jpa，所以必须得手动指定两个Aggregate的Repository，以其中一个为例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@Configuration
public class ProductConfig {

    @Autowired
    private EventStore eventStore;

    @Bean
    @Scope("prototype")
    public ProductAggregate productAggregate(){
        return new ProductAggregate();
    }

    @Bean
    public AggregateFactory<ProductAggregate> productAggregateAggregateFactory(){
        SpringPrototypeAggregateFactory<ProductAggregate> aggregateFactory = new SpringPrototypeAggregateFactory<>();
        aggregateFactory.setPrototypeBeanName("productAggregate");
        return aggregateFactory;
    }

    @Bean
    public Repository<ProductAggregate> productAggregateRepository(){
        EventSourcingRepository<ProductAggregate> repository = new EventSourcingRepository<ProductAggregate>(
                productAggregateAggregateFactory(),
                eventStore
        );
        return repository;
    }
}

使用EventSourcingRepository，必须指定一个AggregateFactory用来反射生成Aggregate的，所以我们这里定义了Aggregate的prototype，并把它注册到AggregateFactory中去。
这样在系统启动时，读取历史Event进行ES还原时，就可以真实再现Aggregate的状态。

配置

前面使用MySQL作为EventStorage是不是感到不爽，那么我们通过引入axon-mongo依赖，使用MongoDB来做EventStorage。
pom的修改我就不写了，着重看下相关配置
先是修改application.property

1
2
3
4
5
6
7
8
# mongo
mongodb.url=10.1.110.24
mongodb.port=27017
# mongodb.username=
# mongodb.password=
mongodb.dbname=axon
mongodb.events.collection.name=events
mongodb.events.snapshot.collection.name=snapshots

通过Spring提供的@Value注解在具体的Configuration类里读取。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@Configuration
public class CommandRepositoryConfiguration {

    @Value("${mongodb.url}")
    private String mongoUrl;

    @Value("${mongodb.dbname}")
    private String mongoDbName;

    @Value("${mongodb.events.collection.name}")
    private String eventsCollectionName;

    @Value("${mongodb.events.snapshot.collection.name}")
    private String snapshotCollectionName;

    @Bean
    public Serializer axonJsonSerializer() {
        return new JacksonSerializer();
    }

    @Bean
    public EventStorageEngine eventStorageEngine(){
        return new MongoEventStorageEngine(
                axonJsonSerializer(),null, axonMongoTemplate(), new DocumentPerEventStorageStrategy());
    }

    @Bean(name = "axonMongoTemplate")
    public MongoTemplate axonMongoTemplate() {
        MongoTemplate template = new DefaultMongoTemplate(mongoClient(), mongoDbName, eventsCollectionName, snapshotCollectionName);
        return template;
    }

    @Bean
    public MongoClient mongoClient(){
        MongoFactory mongoFactory = new MongoFactory();
        mongoFactory.setMongoAddresses(Arrays.asList(new ServerAddress(mongoUrl)));
        return mongoFactory.createMongo();
    }
}

用Jacson做序列化器，MongoClient提供了具体连接实现，MongoTemplate指定了db名称、存放event的collection名称、存放snapshot的collection名称。（snapshot的概念以后再解释）
中间一个参数是做不同版本Event间兼容的，我们先留null。
EventStorageEngine指定MongoEventStorageEngine，spring-boot-autoconfigure中的AxonAutoConfiguration就会帮你把它注入到Axon的配置器中。
这里指的注意的是，使用Jackson做序列化器时，对应的entity的所有需要持久化的field必须都有public getter方法，因为Jackson在反射时默认只读public修饰符的field，否则就会报
com.fasterxml.jackson.databind.JsonMappingException: No serializer found for class com.edi.learn.axon.common.domain.OrderId and no properties discovered to create BeanSerializer (to avoid exception, disable SerializationFeature.FAIL_ON_EMPTY_BEANS) (through reference chain: com.edi.learn.axon.common.events.OrderCreatedEvent[“orderId”])
错误。如果确实不想写，那么在Entity的class声明前加上@JsonAutoDetect(fieldVisibility=JsonAutoDetect.Visibility.ANY)
到此，Command端的实现已基本完成（Event我没写，因为与前文类似），那么我们来看看Query端。

Query端实现

AxonFramework的Query端其实并没有特别的，我们只需要实现一些EventHandler来处理Command端产生的事件，来更新Query端的数据库就行了。
这里我就使用JPA的MySQL实现，spring提供了spring-boot-starter-data-rest，为JPA Repository增加了HateOas风格的REST接口，非常简单，非常方便，堪称无脑。
先定义三个Entity

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@Entity
public class ProductEntry {

  @Id
  private String id;
  @Column
  private String name;
  @Column
  private long price;
  @Column
  private int stock;

  public ProductEntry() {
  }

  public ProductEntry(String id, String name, long price, int stock) {
      this.id = id;
      this.name = name;
      this.price = price;
      this.stock = stock;
  }
  // getter & setter
  ......
}

@Entity
public class OrderEntry {
  @Id
  private String id;
  @Column
  private String username;
  @Column
  private double payment;
  @OneToMany(fetch = FetchType.EAGER, cascade = CascadeType.ALL)
  @JoinColumn(name = "order_id")
  @MapKey(name = "id")
  private Map<String, OrderProductEntry> products;

  public OrderEntry() {
  }

  public OrderEntry(String id, String username, Map<String, OrderProductEntry> products) {
      this.id = id;
      this.username = username;
      this.payment = payment;
      this.products = products;
  }
  // getter & setter
  ......
}

@Entity
public class OrderProductEntry {
  @Id
  @GeneratedValue
  private Long jpaId;
  private String id;
  @Column
  private String name;
  @Column
  private long price;
  @Column
  private int amount;

  public OrderProductEntry() {
  }

  public OrderProductEntry(String id, String name, long price, int amount) {
      this.id = id;
      this.name = name;
      this.price = price;
      this.amount = amount;
  }

  // getter & setter
  ......
}

比较简单，唯一需要注意的是ProductEntry和OrderEntry之间的一对多关系。
然后为它们创建两个Repository

1
2
3
4
@RepositoryRestResource(collectionResourceRel = "orders", path = "orders")
public interface OrderEntryRepository extends PagingAndSortingRepository<OrderEntry, String> {}
@RepositoryRestResource(collectionResourceRel = "products", path = "products")
public interface ProductEntryRepository extends PagingAndSortingRepository<ProductEntry, String> {}

是不是很简单？最后定义handler，为省篇幅，我只写一个

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@Component
public class OrderEventHandler {

    private static final Logger LOGGER = getLogger(OrderEventHandler.class);

    @Autowired
    private OrderEntryRepository repository;

    @EventHandler
    public void on(OrderCreatedEvent event){
        Map<String, OrderProductEntry> map = new HashMap<>();
        event.getProducts().forEach((id, product)->{
            map.put(id,
                    new OrderProductEntry(
                            product.getId(),
                            product.getName(),
                            product.getPrice(),
                            product.getAmount()));
        });
        OrderEntry order = new OrderEntry(event.getOrderId().toString(), event.getUsername(), map);
        repository.save(order);
    }
}

启动类

由于我们使用了axon提供的MongoEventStorageEngine，其内部也使用了JPA，所以我们在启动类还需要把Axon帮我们转Entity的一些类也加到EntityScan中去

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@SpringBootApplication
@ComponentScan(basePackages = {"com.edi.learn"})
@EntityScan(basePackages = {"com.edi.learn",
        "org.axonframework.eventsourcing.eventstore.jpa",
        "org.axonframework.eventhandling.saga.repository.jpa",
        "org.axonframework.eventhandling.tokenstore.jpa"})
@EnableJpaRepositories(basePackages = {"com.edi.learn.axon.query"})
public class Application {

    private static final Logger LOGGER = getLogger(Application.class);

    public static void main(String args[]){
        SpringApplication.run(Application.class, args);
    }
}

启动后，用POST发送请求http://127.0.0.1:8080/product/1?name=ttt&price=10&stock=100 ，查询mongoDB：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
> use axon
> show collections
events
snapshots
system.indexes
> db.events.find().pretty()
{
        "_id" : ObjectId("58dd181073bc0c0fb86d895e"),
        "aggregateIdentifier" : "1",
        "type" : "ProductAggregate",
        "sequenceNumber" : NumberLong(0),
        "serializedPayload" : "{\"id\":\"1\",\"name\":\"ttt\",\"price\":1000,\"stock\":100}",
        "timestamp" : "2017-03-30T14:37:04.075Z",
        "payloadType" : "com.edi.learn.axon.common.events.ProductCreatedEvent",
        "payloadRevision" : null,
        "serializedMetaData" : "{\"traceId\":\"4a298ed4-0d53-402a-ae6b-d79cc5e193bf\",\"correlationId\":\"4a298ed4-0d53-402a-ae6b-d79cc5e193bf\"}",
        "eventIdentifier" : "500f3a8f-7c02-4e8e-bb9c-7b676224ce5c"
}

可以看到生成的EventMessage，与前篇文章中MySQL表里内容基本一致。
再去看下MySQL库的product_entry表，有记录

idnamepricestock
1ttt1000100

用GET请求http://localhost:8080/products 会返回当前所有product信息，加上id http://localhost:8080/products/1 就会返回刚才创建的product。

本篇对应代码：https://github.com/EdisonXu/sbs-axon/tree/master/lesson-4

本文由 EdisonXu - 徐焱飞 创作，采用 CC BY 4.0 CN协议 进行许可。 可自由转载、引用，但需署名作者且注明文章出处。
本文链接为http://edisonxu.com/2017/03/30/axon-cqrs-example.html
如果您觉得文章不错,可以请我喝一杯咖啡！
赏
7 条评论
未登录用户
支持 Markdown 语法
@neveryielding
neveryielding发表于超过 3 年前

公司正打算采用Axon框架重构之前的项目，你这篇教程写得很实用，感谢分享。

@kangkangchen
kangkangchen发表于超过 3 年前

如果使用ES，那么服务启动的时候回加载所有的Aggregation对象吗？感觉不可能是这样啊

@EdisonXu
EdisonXu发表于超过 3 年前

@kangkangchen
如果使用ES，那么服务启动的时候回加载所有的Aggregation对象吗？感觉不可能是这样啊

这个可以自定义，你可以选择在启动时就回溯，加载到内存，也可以在调用时再去回溯，也是很快的。

@Cloud-Leung
Cloud-Leung发表于大约 3 年前

有一个疑问点，就是你在Query端定义了一个EventHandler来接受创建事件，以便在query端保存一份数据供查询使用，但是聚合在每次触发命令的时候都会把历史事件进行回朔重放一边，那query端不是每次在聚合有命令过来时也会收到一大堆命令？但实际上query端只希望获取当前聚合最后的状态就够了，不需要对历史的事件都重演一遍。如果只是创建的话还可以在创建时判断是否是否已经创建过，但如果是状态变化命令的话，就基本上等于query端也跟着跑了一遍。

@EdisonXu
EdisonXu发表于大约 3 年前

@Cloud-Leung
有一个疑问点，就是你在Query端定义了一个EventHandler来接受创建事件，以便在query端保存一份数据供查询使用，但是聚合在每次触发命令的时候都会把历史事件进行回朔重放一边，那query端不是每次在聚合有命令过来时也会收到一大堆命令？但实际上query端只希望获取当前聚合最后的状态就够了，不需要对历史的事件都重演一遍。如果只是创建的话还可以在创建时判断是否是否已经创建过，但如果是状态变化命令的话，就基本上等于query端也跟着跑了一遍。

回溯已保存的历史事件获得Aggregate最新状态时，并不会重新触发EventHandler。

@Cloud-Leung
Cloud-Leung发表于大约 3 年前

@Cloud-Leung
有一个疑问点，就是你在Query端定义了一个EventHandler来接受创建事件，以便在query端保存一份数据供查询使用，但是聚合在每次触发命令的时候都会把历史事件进行回朔重放一边，那query端不是每次在聚合有命令过来时也会收到一大堆命令？但实际上query端只希望获取当前聚合最后的状态就够了，不需要对历史的事件都重演一遍。如果只是创建的话还可以在创建时判断是否是否已经创建过，但如果是状态变化命令的话，就基本上等于query端也跟着跑了一遍。

回溯已保存的历史事件获得Aggregate最新状态时，并不会重新触发EventHandler。

试了一下，确实不会触发外部的EventHandler。
聚合内部自己的EventHandler方法会触发。
是不是说回朔事件的时候只会执行自己聚合内部的EventHandler，外部的不会执行？
还是说axon有其他方式来区分啊？

@pennsong
pennsong发表于超过 2 年前

徐老师您好, 请问

  @EventHandler
    public void on(ProductCreatedEvent event){
        this.id = event.getId();
        this.name = event.getName();
        this.price = event.getPrice();
        this.stock = event.getStock();
        LOGGER.debug("Product [{}] {} {}x{} is created.", id,name,price,stock);
    }

中为何没用@EventSourcingHandler? 谢谢!

文章
Spring Cloud Stream整合Rabbit之重复投递

SpringCloudStream 整合Rabbit 时，消费端在处理失败时，如果需要进行重试，可以有如下几种重试机制： 方法1（默认）：当消费端在处理消息时抛出异常，那么默认会在当前线程的3次的Retry。该方法是默认的，可以通过修改配置文件，指定channel下的参数，例如： 123456789

SpringCloudGateway CORS方案看这篇就够了

问题在Spring Cloud项目中，前后端分离目前很常见，在调试时，会遇到两种情况的跨域： 前端页面通过不同域名或IP访问微服务的后台，例如前端人员会在本地起HttpServer 直连后台开发本地起的服务，此时，如果不加任何配置，前端页面的请求会被浏览器跨域限制拦截，所以，业务服务常常会添加如下代

SpringCloud下skywalking的快速入门

什么是SkyWalking?SkyWalking 是观察性分析平台和应用性能管理系统（APM, Application Performance Management)，由个人开发者 吴晟 于2015年开源，2017年加入Apache孵化器。 它提供分布式跟踪、服务网格遥测分析、度量聚合和可视化的功能

Akka入门系列(八)：akka kafka Consumer

核心API在使用Akka kafka consumer前， 先了解下几个核心API： ConsumerSetting Consumer的配置信息； ConsumerRecord Kafka消息的封装类，包含消息的K、V，以及该消息所属的topic, partition, offset, timest

Akka入门系列(七)：akka kafka Producer

Akka Stream在用Akka去对接Kafka之前，有必要先简单了解下Akka Stream，它是基于Reactive Stream(Akka是其创立成员之一)的。Akka Stream中，将流的拓扑处理逻辑命名为Graph，拓扑中每个操作命名为Operator。它将流式处理抽象为几个核心API

页面
404 Not Found：该页无法显示
徐焱飞

Edison Xu生于84年，标准80后，曾就职于Wipro Shanghai, 爱立信上海研究院，普兰金融，然后进入连续创业期。 技术上，主要从事Java开发，了解C, Scala，写过shell和python脚本。独立开发过安卓应用，尤擅后端开发，小至SSH，大到分布式。设计过底层协议，弄过连接

分类
JavaJava
框架Java/框架
SpringCloudJava/框架/分布式/SpringCloud
分布式Java/框架/分布式
AkkaJava/框架/分布式/Akka
标签
SpringSpring
MQMQ
RabbitRabbit
Spring Cloud StreamSpring-Cloud-Stream
SpringCloudSpringCloud