亳州建设网站,哪些网站教做生物实验,网站推广软件免费版下载,友情链接交换工具一、配置文件
xxxxxx:kafka:bootstrap-servers: xx.xx.xx.xx:9092,xx.xx.xx.xx:9092producer: # 设置大于0的值#xff0c;则客户端会将发送失败的记录重新发送retries: 3 #当有多个消息需要被发送到同一个分区时#xff0c;生产者会把它们放在同一个批次里。该参数指定了一…一、配置文件
xxxxxx:kafka:bootstrap-servers: xx.xx.xx.xx:9092,xx.xx.xx.xx:9092producer: # 设置大于0的值则客户端会将发送失败的记录重新发送retries: 3 #当有多个消息需要被发送到同一个分区时生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小按照字节数计算。16Mbatch-size: 16384linger: 1# 设置生产者内存缓冲区的大小。#32Mbuffer-memory: 33554432# acks0 生产者在成功写入消息之前不会等待任何来自服务器的响应。# acks1 只要集群的首领节点收到消息生产者就会收到一个来自服务器成功响应。# acksall 只有当所有参与复制的节点全部收到消息时生产者才会收到一个来自服务器的成功响应。acks: 1# 指定消息key和消息体的编解码方式 值的序列化方式key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializerconsumer:poll-timeout: 3000key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializerauto-commit: falseoffset-reset: earliestrecords: 10session-timeout: 150000poll-interval: 360000request-timeout: 60000
二、KafkaConfig
package com.xxxxxx.config;import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.*;import java.util.HashMap;
import java.util.Map;Slf4j
Configuration
EnableKafka
public class KafkaConfig {Value(${xxxxxx.kafka.bootstrap-servers})private String servers;Value(${xxxxxx.kafka.producer.retries})private int retries;Value(${xxxxxx.kafka.producer.batch-size})private int batchSize;Value(${xxxxxx-afka.producer.linger})private int linger;Value(${xxxxxx.kafka.producer.buffer-memory})private int bufferMemory;Value(${xxxxxx.kafka.producer.acks})private String acks;Value(${xxxxxx.kafka.producer.key-serializer})private String keyDeserializer;Value(${xxxxxx.kafka.producer.value-serializer})private String valueDeserializer;// 创建生产者配置mapProducerConfig中的可配置属性比spring boot自动配置要多public MapString, Object producerConfigs() {MapString, Object props new HashMap();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);//设置重试次数props.put(ProducerConfig.RETRIES_CONFIG, retries);//达到batchSize大小的时候会发送消息props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);//延时时间延时时间到达之后计算批量发送的大小没达到也发送消息props.put(ProducerConfig.LINGER_MS_CONFIG, linger);//缓冲区的值props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);//序列化手段props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keyDeserializer);props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueDeserializer);props.put(ProducerConfig.ACKS_CONFIG, acks);return props;}public ProducerFactoryString, String producerFactory() {return new DefaultKafkaProducerFactory(producerConfigs());}Bean(name xxxxxxKafkaTemplate)public KafkaTemplateString, String kafkaTemplate() {return new KafkaTemplateString, String(producerFactory());}}三、生产者
Resource(name xxxxxxKafkaTemplate)private KafkaTemplate kafkaTemplate;
kafkaTemplate.send(topic, message);
