使用Spring Boot和Kafka Streams实现CQRS

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

@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.
    }
}

@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);
        }
        
    }

@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);
        }
        
    }

/** * 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;

    }
}

{
    "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"                                                
    }
}