Spring Cloud - 8 (Spring Cloud Stream)

心已赠人 2021-11-16 08:14 223阅读 0赞

# Spring Cloud Stream #

--------------------

# Kafka #

--------------------

**官方网页**

[Apache Kafka][]

## 主要用途 ##

*  消息中间件
 *  流式计算处理
 *  日志

下载地址：[Apache Kafka][Apache Kafka 1]

**执行脚本目录 /bin**

windows 在其单独的目录

## 快速上手 ##

下载并且解压kafka压缩包

运行服务

以Windows为例，首先打开cmd：

**1.  启动zookeeper：**

bin\windows\zookeeper-server-start.bat config\zookeeper.properties

**2.  启动kafka：**

bin\windows\kafka-server-start.bat config\server.properties

### 创建主题topic ###

bin\windows\kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic cloud

cloud

### 生产者：发送消息 ###

\bin\windows>kafka-console-producer.bat --broker-list localhost:9092 --topic cloud
    >hello

### 消费者：接收消息 ###

\bin\windows>kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic cloud --from-beginning
    hello

### 同类产品比较 ###

*  ActiveMQ：JMS（Java Message Service）规范实现
 *  RabbitMQ：AMQP（Advanced Message Queue Protocol）规范实现
 *  Kafka：并非某种规范实现，它的灵活和性能相对是优势

# Spring Kafka #

--------------------

import java.util.Properties;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.Future;
    
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.clients.producer.RecordMetadata;
    import org.apache.kafka.common.serialization.StringSerializer;
    
    public class TestKafka {
    
    	public static void main(String[] args) throws ExecutionException, InterruptedException {
    
    		Properties properties = new Properties();
    
    		properties.setProperty("bootstrap.servers", "localhost:9092");
    		properties.setProperty("key.serializer", StringSerializer.class.getName());
    		properties.setProperty("value.serializer", StringSerializer.class.getName());
    
    		// 创建 Kafka Producer
    		KafkaProducer<String, String> kafkaProducer = new KafkaProducer(properties);
    
    		// 创建Kafka消息 = ProducerRecord
    		String topic = "cloud";
    
    		Integer partition = 0;
    
    		Long timestamp = System.currentTimeMillis();
    
    		String key = "message-key";
    		String value = "how are you!";
    
    		ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, partition, timestamp, key, value);
    
    		// 发送kafka消息
    		Future<RecordMetadata> metadataFuture = kafkaProducer.send(record);
    
    		// 强制执行
    		metadataFuture.get();
    
    	}
    
    }

**官方文档：**[Spring for Apache Kafka][]

### 设计模式 ###

Spring社区对data（Spring-data）操作，有一个基本的模式，**Template**模式：

*  JDBC：JdbcTemplate
 *  Redis：RedisTemplate
 *  Kafka：KafkaTemplate
 *  JMS：JmsTemplate
 *  Rest：RestTemplate

**XXXTemplate一定实现XXXOpeations**

**KafkaTemplate implements KafkaOpeations**

### Maven依赖 ###

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

### 自动装配器： KafkaAutoConfiguration ###

**其中KafkaTemplate会被自动装配：**

import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
    import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.kafka.core.ProducerFactory;
    import org.springframework.kafka.support.ProducerListener;
    
    @Configuration
    public class KafkaAutoConfiguration {
    	
    	 private final KafkaProperties properties;
    
        public KafkaAutoConfiguration(KafkaProperties properties) {
            this.properties = properties;
        }
    	    
    	@Bean
    	@ConditionalOnMissingBean(KafkaTemplate.class)
    	public KafkaTemplate<?,?> kafkaTemplate(ProducerFactory<Object,Object> kafkaProducerFactory,
    	
    	    ProducerListener<Object,Object> kafkaProducerListener){
    	    KafkaTemplate<Object,Object> kafkaTemplate = new KafkaTemplate<Object,Object>(kafkaProducerFactory);
    	
    	    kafkaTemplate.setProducerListener(kafkaProducerListener);
    	    kafkaTemplate.setDefaultTopic(this.properties.getTemplate().getDefaultTopic());
    	
    	    return kafkaTemplate;
    	}
    }

### 创建生产者 ###

**增加生产者配置**

**application.properties**

>    全局配置：
> 
>     \#\#\# Kafka生产者配置     
> 
>     spring.kafka.producer.bootstrapServers = localhost:9092

### Kafka生产者配置
    
    # spring.kafka.producer.bootstrapServers = localhost:9092
    
    spring.kafka.producer.keySerializer = org.apache.kafka.common.serialization.StringSerializer
    
    spring.kafka.producer.valueSerializer = org.apache.kafka.common.serialization.StringSerializer

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;
    
    @RestController
    public class KafkaProducerController {
    
    	private final KafkaTemplate<String, String> kafkaTemplate;
    
    	private final String topic;
    
    	@Autowired
    	public KafkaProducerController(KafkaTemplate<String, String> kafkaTemplate, @Value("${kafka.topic}") String topic) {
    
    		this.kafkaTemplate = kafkaTemplate;
    		this.topic = topic;
    	}
    
    	@PostMapping("/message/send")
    	public Boolean sendMessage(@RequestParam(required=false)String message) {
    		kafkaTemplate.send(topic, message);
    		return true;
    	}
    
    }

### 创建消费者 ###

**增加消费者配置**

### Kafka消费者配置
    spring.kafka.consumer.groupId = cloud-1
    
    spring.kafka.consumer.keyDeserializer = org.apache.kafka.common.serialization.StringDeserializer
    
    spring.kafka.consumer.valueDeserializer = org.apache.kafka.common.serialization.StringDeserializer

import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    
    @Component
    public class KafkaConsumerListener {
    
    	@KafkaListener(topics = "${kafka.topic}")
    	public void onMessage(String message) {
    
    		System.out.print("kafka 消费者监听器，接收到消息：" + message);
    
    	}
    }

# Spring Cloud Stream #

--------------------

**基本概念**

*  Source：来源，近义词：Producer、Publisher
 *  Sink：接收器，近义词：Consumer、Subscriber
 *  Processor：对于上流而言是Sink，对于下流而言是Source

**Reactive Streams：**

*  Publisher
 *  Subscriber
 *  Processor

# Spring Cloud Stream Binder:kafka #

### 定义标准消息发送源 ###

@Component
    @EnableBinding(Source.class)
    public class MessageProducerBean {
    
        @Autowired
        @Qualifier(Source.OUTPUT)
        private MessageChannel messageChannel; 
        
        @Autowired
        private Source source;
        
        
        public void send(String message) {
            messageChannel.send(MessageBuilder.withPayload(message).build());
            source.output().send(MessageBuilder.withPayload(message).build());
        }
    
    }

### 注入 ###

@RestController
    public class KafkaProducerController {
    
        private final KafkaTemplate<String, String> kafkaTemplate;
    
        private final MessageProducerBean messageProducerBean;
    
        private final String topic;
    
        @Autowired
        public KafkaProducerController(KafkaTemplate<String, String> kafkaTemplate, 
                @Value("${kafka.topic}") String topic,
                MessageProducerBean messageProducerBean) {
    
            this.kafkaTemplate = kafkaTemplate;
            this.messageProducerBean = messageProducerBean;
            this.topic = topic;
        }
    
        @PostMapping("/message/send")
        public Boolean sendMessage(@RequestParam(required=false)String message) {
            kafkaTemplate.send(topic, message);
            return true;
        }
        
        @GetMapping("/message/output/send")
        public Boolean outputSend(@RequestParam String message) {
            messageProducerBean.send(message);
            return true;
        }    
        
    
    }

### 实现标准Sink监听 ###

@Component
    @EnableBinding(Sink.class)
    public class MessageConsumerBean {
    
        @Autowired
        @Qualifier(Sink.INPUT)
        private SubscribableChannel subscribableChannel;
        
        @Autowired
        private Sink sink;
    
        @PostConstruct
        public void init() {
            subscribableChannel.subscribe(new MessageHandler() {
                
                @Override
                public void handleMessage(Message<?> message) throws MessagingException {
                    System.out.println("init: "+message.getPayload());
                    
                }
            });
        }
        
        @ServiceActivator(inputChannel = Sink.INPUT)
        public void serviceActivator(Object message) {
            System.out.println("serviceActivator: "+message);
        }
        
        @StreamListener(Sink.INPUT)
        public void streamListener(String message) {
            System.out.println("streamListener: "+message);
        }
        
    }

### 自定义标准消息发送源 ###

public interface MySource {
        
        String MYOUTPUT = "myoutput";
    
        
        @Output(MySource.MYOUTPUT)
        MessageChannel myoutput();
    
    }

@Component
    @EnableBinding(MySource.class)
    public class MyMessageProducerBean {
    
    	@Autowired
    	@Qualifier(MySource.MYOUTPUT)
    	private MessageChannel messageChannel; 
    	
    	@Autowired
    	private MySource mySource;
    	
    	
    	public void send(String message) {
    		messageChannel.send(MessageBuilder.withPayload(message).build());
    		mySource.myoutput().send(MessageBuilder.withPayload(message).build());
    	}
    
    }

### 自定义Sink监听 ###

public interface MySink {
    
        String MYINPUT = "myinput";
    
        @Input(MySink.MYINPUT)
        SubscribableChannel input();
    
    }

@Component
    @EnableBinding(MySink.class)
    public class MyMessageConsumerBean {
    
        @Autowired
        @Qualifier(MySink.MYINPUT)
        private SubscribableChannel subscribableChannel;
        
        @Autowired
        private MySink mySink;
    
        @PostConstruct
        public void init() {
            subscribableChannel.subscribe(new MessageHandler() {
                
                @Override
                public void handleMessage(Message<?> message) throws MessagingException {
                    System.out.println("my - Sink: "+message.getPayload());
                    
                }
            });
        }
        
        @ServiceActivator(inputChannel = MySink.MYINPUT)
        public void serviceActivator(Object message) {
            System.out.println("my - serviceActivator: "+message);
        }
        
        @StreamListener(MySink.MYINPUT)
        public void streamListener(String message) {
            System.out.println("my - streamListener: "+message);
        }
        
    }

### 配置项 ###

kafka.topic.my = mytopic
    spring.cloud.stream.bindings.myoutput.destination=${kafka.topic.my}
    spring.cloud.stream.bindings.myinput.destination=${kafka.topic.my}

# Spring Cloud Stream Binder:rabbit #

--------------------

**重构Kafka工程，删除强依赖**

Stream-kafka实现源码：https://pan.baidu.com/s/1RX5W2wMj4h\_SKDkjlPQHkA 提取码：lwak

Stream-rabbit实现源码：https://pan.baidu.com/s/1AX6asvmATN9-dYrhIIfS7w 提取码：dsc5

[Apache Kafka]: http://kafka.apache.org/
[Apache Kafka 1]: http://kafka.apache.org/downloads
[Spring for Apache Kafka]: https://docs.spring.io/spring-kafka/reference/html/