Kafka及Spring Cloud Stream

        <dependency>            <groupId>org.springframework.kafka</groupId>            <artifactId>spring-kafka</artifactId>        </dependency>

package com.example.springkafka.api;import org.apache.kafka.clients.producer.KafkaProducer;import org.apache.kafka.clients.producer.ProducerRecord;import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties;/** * @Date: 2018/11/6 20:25 * @Description: 生产者 */public class KafkaProducerDemo {    public static void main(String[] args) {        Properties properties = new Properties();        properties.setProperty("bootstrap.servers","192.168.3.221:9092");        properties.setProperty("key.serializer", StringSerializer.class.getName());        properties.setProperty("value.serializer", StringSerializer.class.getName());        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);        String topic = "message"; // 主题        Integer partition = 0; // 指定分区        long timeMillis = System.currentTimeMillis(); // 毫秒值 15分钟        String key = "key-message"; // key        String value = "value-message"; // value        // 创建ProducerRecord        ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(topic, partition, timeMillis, key, value);        // 生产消息        kafkaProducer.send(producerRecord);        kafkaProducer.close();    }}

package com.example.springkafka.api;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.apache.kafka.common.serialization.StringDeserializer;import java.util.Arrays;import java.util.Properties;/** * @Date: 2018/11/6 20:25 * @Description: 消费者 */public class KafkaConsumerDemo {    public static void main(String[] args) {        Properties properties = new Properties();        properties.setProperty("bootstrap.servers", "192.168.3.221:9092");        properties.setProperty("group.id", "group-1");        properties.setProperty("key.deserializer", StringDeserializer.class.getName());        properties.setProperty("value.deserializer", StringDeserializer.class.getName());        // 创建kafka的消费者对象        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);        // 订阅kafka主题        kafkaConsumer.subscribe(Arrays.asList("message"));        while (true) {            ConsumerRecords<String, String> records = kafkaConsumer.poll(1000);            for (ConsumerRecord<String, String> record : records)                System.out.printf("========offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());        }    }}

        <dependency>            <groupId>org.springframework.cloud</groupId>            <artifactId>spring-cloud-stream-binder-kafka</artifactId>        </dependency>

# 生成者配置spring:  kafka:    producer:      key-serializer: org.apache.kafka.common.serialization.StringSerializer      value-serializer: org.apache.kafka.common.serialization.StringSerializer    bootstrap-servers: 192.168.3.221:9092    consumer: # 消费者      group-id: gerry-1      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer      value-deserializer: org.apache.kafka.common.serialization.StringDeserializerkafka:  topic: gerry

package com.example.springcloudkafka.controller;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.beans.factory.annotation.Value;import org.springframework.kafka.core.KafkaTemplate;import org.springframework.web.bind.annotation.PostMapping;import org.springframework.web.bind.annotation.RequestParam;import org.springframework.web.bind.annotation.RestController;/** * @Date: 2018/11/6 21:03 * @Description: */@RestControllerpublic class KafkaProducerController {    public final KafkaTemplate<String, String> kafkaTemplate;    private final String topic;    public KafkaProducerController(KafkaTemplate<String, String> kafkaTemplate,                                   @Value("${kafka.topic}") String topic) {        this.kafkaTemplate = kafkaTemplate;        this.topic = topic;    }    @PostMapping("message/send") // 这种方式只支持post    public boolean sendMessage(@RequestParam String message) {        kafkaTemplate.send(topic,message);        return true;    }}

package com.example.springcloudkafka.listener;import org.springframework.kafka.annotation.KafkaListener;import org.springframework.stereotype.Component;/** * @Date: 2018/11/6 21:20 * @Description: */@Componentpublic class KafkaConsumerListener {    @KafkaListener(topics={"${kafka.topic}"})    public void getMessage(String message) {        System.out.println("kafka 消费者监听，接收到消息:" + message);    }}

        <dependency>            <groupId>org.springframework.cloud</groupId>            <artifactId>spring-cloud-stream-binder-kafka</artifactId>        </dependency>

# 生成者配置spring:  kafka:    bootstrap-servers: 192.168.3.221:9092  cloud:    stream:      bindings:        output:          destination: ${kafka.topic}        input:          destination: ${kafka.topic}kafka:  topic: cloud-stream

package com.example.springcloudstreamkafkademo.producer;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.beans.factory.annotation.Qualifier;import org.springframework.cloud.stream.annotation.EnableBinding;import org.springframework.cloud.stream.messaging.Source;import org.springframework.messaging.MessageChannel;import org.springframework.messaging.support.MessageBuilder;import org.springframework.stereotype.Component;@Component@EnableBinding(Source.class)public class MessageProducerBean {    @Autowired    @Qualifier(Source.OUTPUT)    private MessageChannel messageChannel;    @Autowired    private Source source;    /**     * 发送信息     * @param message     */    public void send(String message) {        // 通过消息管道发送消息        // messageChannel.send(MessageBuilder.withPayload(message).build());        source.output().send(MessageBuilder.withPayload(message).build());    }}

package com.example.springcloudstreamkafkademo.consumer;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.beans.factory.annotation.Qualifier;import org.springframework.cloud.stream.annotation.EnableBinding;import org.springframework.cloud.stream.annotation.StreamListener;import org.springframework.cloud.stream.messaging.Sink;import org.springframework.integration.annotation.ServiceActivator;import org.springframework.messaging.SubscribableChannel;import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;@Component@EnableBinding(value={Sink.class})public class MessageConsumerBean {    @Autowired    @Qualifier(Sink.INPUT)    private SubscribableChannel subscribableChannel;    //1、 当subscribableChannel注入完成后完成回调    @PostConstruct    public void init() {        subscribableChannel.subscribe(message->{            System.out.println(message.getPayload());        });    }    // 2、@ServiceActivator    @ServiceActivator(inputChannel=Sink.INPUT)    public void message(String message) {        System.out.println("@ServiceActivator:"+message);    }    //3、@StreamListener    @StreamListener(Sink.INPUT)    public void onMessage(String message) {        System.out.println("@StreamListener:"+message);    }}

        <dependency>            <groupId>org.springframework.cloud</groupId>            <artifactId>spring-cloud-stream-binder-rabbit</artifactId>        </dependency>

spring:  cloud:    stream:      bindings:        output:          destination: ${rabbit.queue}        input:          destination: ${rabbit.queue}  rabbitmq:    host: 192.168.3.221    port: 5672    username: rabbit    password: rabbitrabbit:  queue: cloud-stream-queue