使用Spring Boot和Kafka Streams實現CQRS

ApacheKafka SpringBoot · 發表 2018-10-21 08:54:00

摘要：本文是David Romero一篇Spring + Kafka Stream實現CQRS的案例程式碼：去年九月，我的同事伊萬·古鐵雷斯和我談到我們cowokers如何實現事件與Kafka Stream，我開發了一個Kafka Stream，它讀取包含來自Twitter的“Java”字樣的...

本文是David Romero一篇Spring + Kafka Stream實現CQRS的案例程式碼：

去年九月，我的同事伊萬·古鐵雷斯和我談到我們cowokers如何實現事件與Kafka Stream，我開發了一個Kafka Stream，它讀取包含來自Twitter的“Java”字樣的推文，按使用者名稱分組推文，並選擇最喜歡的推文。管道結束端將重新選擇的資訊再發送到PostgreSQL

由於我們收到了積極的反饋，並且我們學到了很多東西，所以我想分享這個演示，以便任何想要看一眼的人都可以使用它。

2.實施

該演示是基於Kafka和Kafka Streams 的CQRS模式的實現。Kafka能夠解耦read（Query）和write（Command）操作，這有助於我們更快地開發事件源應用。

堆疊

整個堆疊已在Docker中實現，因為它集成了多個工具及其隔離級別時的簡單性。堆疊由

version: '3.1'
services:

#############
# Kafka
#############
zookeeper:
image: confluentinc/cp-zookeeper
container_name: zookeeper
network_mode: host
ports:
- "2181:2181"
environment:
ZOOKEEPER_CLIENT_PORT: 2181

kafka:
image: confluentinc/cp-kafka
container_name: kafka
network_mode: host
depends_on:
- zookeeper
ports:
- "9092:9092"
environment:
KAFKA_ZOOKEEPER_CONNECT: localhost:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 # We have only 1 broker, so offsets topic can only have one replication factor.

connect:
image: confluentinc/cp-kafka-connect
container_name: kafka-connect
network_mode: host
ports:
- "8083:8083"
depends_on:
- zookeeper
- kafka
volumes:
- $PWD/connect-plugins:/etc/kafka-connect/jars # in this volume is located the postgre driver.
environment:
CONNECT_BOOTSTRAP_SERVERS: kafka:9092
CONNECT_REST_PORT: 8083 # Kafka connect creates an endpoint in order to add connectors
CONNECT_REST_ADVERTISED_HOST_NAME: "kafka-connect"
CONNECT_GROUP_ID: kafka-connect
CONNECT_ZOOKEEPER_CONNECT: zookeeper:2181
CONNECT_CONFIG_STORAGE_TOPIC: kafka-connect-config
CONNECT_OFFSET_STORAGE_TOPIC: kafka-connect-offsets
CONNECT_STATUS_STORAGE_TOPIC: kafka-connect-status
CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1 # We have only 1 broker, so we can only have 1 replication factor.
CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1 
CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1 
CONNECT_KEY_CONVERTER: "org.apache.kafka.connect.storage.StringConverter" # We receive a string as key and a json as value
CONNECT_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.storage.StringConverter"
CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
CONNECT_PLUGIN_PATH: /usr/share/java,/etc/kafka-connect/jars

#############
# PostgreSQL
#############
db:
container_name: postgresql
network_mode: host
image: postgres
restart: always
ports:
- "5432:5432"
environment:
POSTGRES_DB: influencers
POSTGRES_USER: user
POSTGRES_PASSWORD: 1234

上面配置在ofollow,noindex" target="_blank">docker-compose 檔案中，也包含此演示中涉及的所有工具：

1. Zookeper：卡夫卡不可分割的合作伙伴。

2. Kafka：主要角色。你需要設定zookeeper ip。

3. Kafka Connector：4個主要的Kafka核心API之一。它負責讀取所提供topic的記錄並將其插入PostgreSQL。

4. PostgreSQL：SQL資料庫。

生產者

這是寫入資料到Kafka的應用程式。我們的基礎設施負責閱讀Twitter中包含“Java”字樣的推文並將其傳送給Kafka。

以下程式碼包含兩個部分：Twitter Stream和Kafka Producer。

Twitter Stream：建立推文資料流。如果要在消費之前過濾流，也可以新增FilterQuery。您需要憑據才能訪問Twitter API。

Kafka Producer：它將記錄傳送給kafka。在我們的演示中，它將沒有key的記錄傳送到“推文”主題。

@SpringBootApplication
@Slf4j
public class DemoTwitterKafkaProducerApplication {

public static void main(String[] args) {

// Kafka config
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); //Kafka cluster hosts.
properties.put(ProducerConfig.CLIENT_ID_CONFIG, "demo-twitter-kafka-application-producer"); // Group id
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
Producer<String, String> producer = new KafkaProducer<>(properties);

// Twitter Stream
final TwitterStream twitterStream = new TwitterStreamFactory().getInstance();
final StatusListener listener = new StatusListener() {
public void onStatus(Status status) {
final long likes = getLikes(status);
final String tweet = status.getText();
final String content = status.getUser().getName() + "::::" + tweet + "::::" + likes;
log.info(content);
producer.send(new ProducerRecord<>("tweets", content));
}

//Some methods have been omitted for simplicity.

private long getLikes(Status status) {
return status.getRetweetedStatus() != null ? status.getRetweetedStatus().getFavoriteCount() : 0; // Likes can be null.
}
};
twitterStream.addListener(listener);
final FilterQuery tweetFilterQuery = new FilterQuery();
tweetFilterQuery.track(new String[] { "Java" });
twitterStream.filter(tweetFilterQuery);
SpringApplication.run(DemoTwitterKafkaProducerApplication.class, args);
Runtime.getRuntime().addShutdownHook(new Thread(() -> producer.close())); //Kafka producer should close when application finishes.
}
}

這個應用程式是一個Spring Boot應用程式。

Kafka Stream

我們基礎設施的主要部分負責從“tweets”主題topic中閱讀推文，按使用者名稱分組，計算推文，提取最喜歡的推文並將其傳送給“influencers”的新主題。

讓我們關注下一個程式碼塊的兩個最重要的方法：

1. 流方法：Kafka Stream Java API遵循與Java 8 Stream API相同的命名法。在管道中執行的第一個操作是選擇金鑰，因為每次金鑰改變時，在主題中執行重新分割槽操作。所以，我們應該儘可能少地改變金鑰。然後，我們必須計算大多數人喜歡積累的推文。並且由於此操作是有狀態操作，我們需要執行聚合。聚合操作將在以下專案中詳述。最後，我們需要將記錄傳送到名為“influencers”的輸出主題。對於此任務，我們需要將Influencer類對映到InfluencerJsonSchema類，然後使用to方法。InfluencerJsonSchema類將在Kafka Connector部分中解釋。Peek方法用於除錯目的。

2.aggregateInfoToInfluencer方法：這是一個有狀態操作。收到三個引數：使用者名稱，主題的原始推文和之前儲存的Influencer。在推文計數器中新增一個，並將喜歡與喜歡的推文進行比較。返回Influecer類的新例項，以保持不變性。

@Configuration
@EnableKafkaStreams
static class KafkaConsumerConfiguration {

final Serde<Influencer> jsonSerde = new JsonSerde<>(Influencer.class);
final Materialized<String, Influencer, KeyValueStore<Bytes, byte[]>> materialized = Materialized.<String, Influencer, KeyValueStore<Bytes, byte[]>>as("aggregation-tweets-by-likes").withValueSerde(jsonSerde);

@Bean
KStream<String, String> stream(StreamsBuilder streamBuilder){
final KStream<String, String> stream = streamBuilder.stream("tweets");
stream
.selectKey(( key , value ) -> String.valueOf(value.split("::::")[0]))
.groupByKey()
.aggregate(Influencer::init, this::aggregateInfoToInfluencer, materialized)
.mapValues(InfluencerJsonSchema::new)
.toStream()
.peek( (username, jsonSchema) -> log.info("Sending a new tweet from user: {}", username))
.to("influencers", Produced.with(Serdes.String(), new JsonSerde<>(InfluencerJsonSchema.class)));
return stream;
}

private Influencer aggregateInfoToInfluencer(String username, String tweet, Influencer influencer) {
final long likes = Long.valueOf(tweet.split("::::")[2]);
if ( likes >= influencer.getLikes() ) {
return new Influencer(influencer.getTweets()+1, username, String.valueOf(tweet.split("::::")[1]), likes);
} else {
return new Influencer(influencer.getTweets()+1, username, influencer.getContent(), influencer.getLikes());
}
}

@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public StreamsConfig kStreamsConfigs(KafkaProperties kafkaProperties) {
Map<String, Object> props = new HashMap<>();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "demo-twitter-kafka-application");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
return new StreamsConfig(props);
}

}

@EnableKafkaStreams註釋和kStreamsConfigs方法負責將Kafka Stream API與Spring Framework整合。

在上面的程式碼塊中提到了Influencer 類，為了便於閱讀，這裡提供了影響者類的程式碼：

@RequiredArgsConstructor
 @Getter
public static class Influencer {

final long tweets;

final String username;

final String content;

final long likes;

static Influencer init() {
return new Influencer(0, "","", 0);
}

@JsonCreator
static Influencer fromJson(@JsonProperty("tweets") long tweetCounts, @JsonProperty("username") String username, @JsonProperty("content") String content, @JsonProperty("likes") long likes) {
return new Influencer(tweetCounts, username, content, likes);
}

}

fromJson由於Kafka Stream使用了其序列化方法，該方法是Kafka強制性要求的。如果您想了解有關此主題的更多資訊，請參閱Kafka Stream Serde。

這個應用程式是一個Spring Boot應用程式。

Kafka Connector

一旦我們將資料餵給了我們的新主題“influencers”，我們就必須將資料儲存到Postgre。對於此任務，Kafka提供了一個名為Kafka Connect的強大API 。由Apache Kafka開發人員建立的Confluent公司為許多第三方工具開發了多個聯結器。對於JDBC，推出兩個聯結器：source和sink。

1. source源聯結器是從jdbc驅動程式讀取資料並將資料傳送到Kafka。

2. sink接收器聯結器從Kafka讀取資料並將其傳送到jdbc驅動程式。

我們將使用JDBC Sink聯結器，此聯結器需要schema資訊才能將主題記錄對映到sql記錄。在我們的演示中，schema 在主題的記錄中提供。因此，我們必須在資料管道中將Influecer類對映到InfluencerJsonSchema類。

在以下程式碼中，您可以看到schema 的傳送方式。

/** * https://gist.github.com/rmoff/2b922fd1f9baf3ba1d66b98e9dd7b364 **/
@Getter
public class InfluencerJsonSchema {

Schema schema;
Influencer payload;

InfluencerJsonSchema(long tweetCounts, String username, String content, long likes) {
this.payload = new Influencer(tweetCounts, username, content, likes);
Field fieldTweetCounts = Field.builder().field("tweets").type("int64").build();
Field fieldContent = Field.builder().field("content").type("string").build();
Field fieldUsername = Field.builder().field("username").type("string").build();
Field fieldLikes = Field.builder().field("likes").type("int64").build();
this.schema = new Schema("struct", Arrays.asList(fieldUsername,fieldContent,fieldLikes,fieldTweetCounts));
}

public InfluencerJsonSchema(Influencer influencer) {
this(influencer.getTweets(),influencer.getUsername(),influencer.getContent(),influencer.getLikes());
}

@Getter
@AllArgsConstructor
static class Schema {

String type;
List<Field> fields;

}

@Getter
@Builder
static class Field {

String type;
String field;

}
}

然後，我們需要配置我們的Kafka聯結器。應提供源主題，目標表，主鍵或URL連線。特別提到'insert.mode'欄位。我們使用'upsert'模式，因為主鍵是使用者名稱，因此將根據使用者之前是否已經持久來插入或更新記錄。

{
"name": "jdbc-sink",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"tasks.max": "1",
"topics": "influencers",
"table.name.format": "influencer",
"connection.url": "jdbc:postgresql://postgresql:5432/influencers?user=user&password=1234",
"auto.create": "true",
"insert.mode": "upsert",
"pk.fields": "username",
"pk.mode": "record_key"
}
}

上面的json程式碼已儲存到一個檔案中，以便對其進行跟進

一旦我們開發了聯結器，我們就必須將聯結器新增到我們的Kafka Connector容器中，這可以通過簡單的curl命令來執行。

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d @connect-plugins/jdbc-sink.json

消費閱讀者

我們開發了一個簡單的Spring Boot應用程式來讀取Postgre中插入的記錄。這個應用程式非常簡單，程式碼將被跳過這篇文章，因為它沒關係。

如果你想看到程式碼，可以在 Github中找到

3.如何執行？

如果要執行演示，則必須執行以下命令。

1. docker-compose up

2. curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d @connect-plugins/jdbc-sink.json

3. mvn clean spring-boot:run -pl producer

4. mvn clean spring-boot:run -pl consumer

5. mvn clean spring-boot:run -pl reader

結論

這個演示向我們展示了CQRS實現的一個很好的例子，以及使用Kafka實現這種模式是多麼容易。

在我看來，Kafka Stream是Kafka最強大的API，因為它提供了一個簡單的API，具有很棒的功能，可以從所有必要的實現中抽象出來，消耗來自Kafka的記錄，並允許您專注於開發用於管理大資料流的強大管道。

此外，Spring Framework提供了額外的抽象層，允許我們將Kafka與Spring Boot應用程式整合。

提供了本文的完整原始碼

使用Spring Boot和Kafka Streams實現CQRS

您可能也會喜歡…