kafka生产者Producer、消费者Consumer的拦截器interceptor |
您所在的位置：网站首页 › kafka中producerConfig › kafka生产者Producer、消费者Consumer的拦截器interceptor |
|
1、Producer的拦截器interceptor，和consumer端的拦截器interceptor是在kafka0.10版本被引入的，主要用于实现clients端的定制化控制逻辑，生产者拦截器可以用在消息发送前做一些准备工作，使用场景，如下所示： 1）、按照某个规则过滤掉不符合要求的消息。 2）、修改消息的内容。 3）、统计类需求。 代码语言：javascript复制 1 package com.demo.kafka.listener; 2 3 import java.util.Map; 4 5 import org.apache.kafka.clients.producer.ProducerInterceptor; 6 import org.apache.kafka.clients.producer.ProducerRecord; 7 import org.apache.kafka.clients.producer.RecordMetadata; 8 9 /** 10 * 生产者拦截器 11 * 12 * @author 生产者拦截器 13 * 14 */ 15 16 public class ProducerInterceptorPrefix implements ProducerInterceptor { 17 18 // 发送成功计数 19 private volatile long sendSuccess = 0; 20 21 // 发送失败计数 22 private volatile long sendFailure = 0; 23 24 /** 25 * 26 */ 27 @Override 28 public void configure(Map configs) { 29 30 } 31 32 /** 33 * 发送消息已经操作消息的方法 34 */ 35 @Override 36 public ProducerRecord onSend(ProducerRecord record) { 37 String modifiedValue = "前缀prefix : " + record.value(); 38 ProducerRecord producerRecord = new ProducerRecord( 39 record.topic(), // 主题 40 record.partition(), // 分区 41 record.timestamp(), // 时间戳 42 record.key(), // key值 43 modifiedValue, // value值 44 record.headers()); // 消息头 45 return producerRecord; 46 } 47 48 /** 49 * ack确认的方法 50 */ 51 @Override 52 public void onAcknowledgement(RecordMetadata metadata, Exception exception) { 53 if(exception == null) { 54 sendSuccess++; 55 }else { 56 sendFailure++; 57 } 58 } 59 60 /** 61 * 关闭的方法，发送成功之后会将拦截器关闭，调用此方法 62 */ 63 @Override 64 public void close() { 65 double successRation = (double)sendSuccess / (sendSuccess + sendFailure); 66 System.out.println("【INFO 】 发送成功率： " + String.format("%f", successRation * 100) + "%"); 67 } 68 69 }生产者客户端要配置一下Producer的拦截器interceptor，如下所示： 代码语言：javascript复制 1 package com.demo.kafka.producer; 2 3 import java.util.Properties; 4 import java.util.concurrent.ExecutionException; 5 6 import org.apache.kafka.clients.producer.KafkaProducer; 7 import org.apache.kafka.clients.producer.ProducerConfig; 8 import org.apache.kafka.clients.producer.ProducerRecord; 9 import org.apache.kafka.clients.producer.RecordMetadata; 10 import org.apache.kafka.common.serialization.StringSerializer; 11 12 import com.demo.kafka.listener.ProducerInterceptorPrefix; 13 14 public class KafkaProducerSimple { 15 16 // 设置服务器地址 17 private static final String brokerList = "192.168.110.142:9092"; 18 19 // 设置主题 20 private static final String topic = "topic-demo"; 21 22 public static void main(String[] args) { 23 Properties properties = new Properties(); 24 // 设置key的序列化器 25 properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); 26 27 // 设置重试次数 28 properties.put(ProducerConfig.RETRIES_CONFIG, 10); 29 30 // 设置值的序列化器 31 properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); 32 33 // 打印输出序列化器的路径信息 34 System.err.println(StringSerializer.class.getName()); 35 36 // 设置集群地址 37 properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList); 38 39 // 自定义拦截器使用，可以计算发送成功率或者失败率，进行消息的拼接或者过滤操作 40 properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerInterceptorPrefix.class.getName()); 41 42 // 将参数配置到生产者对象中 43 KafkaProducer producer = new KafkaProducer(properties); 44 45 for (int i = 0; i < 100000; i++) { 46 // 生产者消息记录 47 ProducerRecord record = new ProducerRecord(topic, "hello world!!!" + i); 48 // 同步获取消息 49 // RecordMetadata recordMetadata = producer.send(record).get(); 50 producer.send(record); 51 } 52 53 // 关闭 54 producer.close(); 55 } 56 57 }消费者代码，如下所示： 代码语言：javascript复制 1 package com.demo.kafka.consumer; 2 3 import java.time.Duration; 4 import java.util.Collections; 5 import java.util.Properties; 6 7 import org.apache.kafka.clients.consumer.ConsumerConfig; 8 import org.apache.kafka.clients.consumer.ConsumerRecord; 9 import org.apache.kafka.clients.consumer.ConsumerRecords; 10 import org.apache.kafka.clients.consumer.KafkaConsumer; 11 import org.apache.kafka.clients.producer.ProducerConfig; 12 import org.apache.kafka.common.serialization.StringDeserializer; 13 14 public class KafkaConsumerSimple { 15 16 // 设置服务器地址 17 private static final String bootstrapServer = "192.168.110.142:9092"; 18 19 // 设置主题 20 private static final String topic = "topic-demo"; 21 22 // 设置消费者组 23 private static final String groupId = "group.demo"; 24 25 public static void main(String[] args) { 26 Properties properties = new Properties(); 27 // 设置反序列化key参数信息 28 properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); 29 // 设置反序列化value参数信息 30 properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); 31 32 // 设置服务器列表信息 33 properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer); 34 35 // 设置消费者组信息 36 properties.put("group.id", groupId); 37 38 // 将参数设置到消费者参数中 39 KafkaConsumer consumer = new KafkaConsumer(properties); 40 41 // 消息订阅 42 consumer.subscribe(Collections.singletonList(topic)); 43 44 while (true) { 45 // 每隔一秒监听一次 46 ConsumerRecords records = consumer.poll(Duration.ofMillis(1000)); 47 // 获取到消息信息 48 for (ConsumerRecord record : records) { 49 System.err.println(record.toString()); 50 } 51 } 52 53 } 54 55 }2、生产者的acks参数，这个参数用来指定分区中必须有多少副本来收到这条消息，之后生产者才会认为这条消息写入成功的。acks是生产者客户端中非常重要的一个参数，它涉及到消息的可靠性和吞吐量之间的权衡。 1）、ack等于0，生产者在成功写入消息之前不会等待任何来自服务器的响应。如果出现问题生产者是感知不到的，消息就丢失了，不过因为生产者不需要等待服务器响应，所以他可以以网络能够支持的最大速度发送消息，从而达到很高的吞吐量。 2）、acks等于1，默认值为1，只要集群的首领节点收到消息，生产者就会收到一个来自服务器的成功响应。如果消息无法达到首领节点，比如首领节点崩溃，新的首领节点还没有被选举出来，生产者会收到一个错误响应，为了避免数据丢失，生产者会重发消息。但是这样还有可能会导致数据丢失，如果收到写成功通知，此时首领节点还没有来的及同步数据到follower节点，首领节点崩溃，就会导致数据丢失。 3）、acks等于-1，只有当所有参与复制的节点收到消息时候，生产者会收到一个来自服务器额成功响应，这种模式 最安全的，他可以保证不止一个服务器收到消息。 注意，acks参数配置的是一个字符串类型，而不是整数类型，如果配置为整数类型会抛出异常信息。 3、kafka消费者订阅主题和分区，创建完消费者后我们便可以订阅主题了，只需要调用subscribe方法即可，这个方法会接受一个主题列表，如下所示： 另外，我们也可以使用正则表达式来匹配多个主题，而且订阅之后如果又有匹配的新主题，那么这个消费组立即对其进行消费。正则表达式在连接kafka与其他系统非常有用。比如订阅所有的测试主题。 代码语言：javascript复制 1 package com.demo.kafka.consumer; 2 3 import java.time.Duration; 4 import java.util.Arrays; 5 import java.util.Collections; 6 import java.util.Properties; 7 import java.util.regex.Pattern; 8 9 import org.apache.kafka.clients.consumer.ConsumerConfig; 10 import org.apache.kafka.clients.consumer.ConsumerRecord; 11 import org.apache.kafka.clients.consumer.ConsumerRecords; 12 import org.apache.kafka.clients.consumer.KafkaConsumer; 13 import org.apache.kafka.clients.producer.ProducerConfig; 14 import org.apache.kafka.common.TopicPartition; 15 import org.apache.kafka.common.serialization.StringDeserializer; 16 17 public class KafkaConsumerSimple { 18 19 // 设置服务器地址 20 private static final String bootstrapServer = "192.168.110.142:9092"; 21 22 // 设置主题 23 private static final String topic = "topic-demo"; 24 25 // 设置主题 26 private static final String topic2 = "topic-demo2"; 27 28 // 设置消费者组 29 private static final String groupId = "group.demo"; 30 31 public static void main(String[] args) { 32 Properties properties = new Properties(); 33 // 设置反序列化key参数信息 34 properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); 35 // 设置反序列化value参数信息 36 properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); 37 38 // 设置服务器列表信息，必填参数，该参数和生产者相同，，制定链接kafka集群所需的broker地址清单，可以设置一个或者多个 39 properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer); 40 41 // 设置消费者组信息，消费者隶属的消费组，默认为空，如果设置为空，则会抛出异常，这个参数要设置成具有一定业务含义的名称 42 properties.put("group.id", groupId); 43 44 // 制定kafka消费者对应的客户端id，默认为空，如果不设置kafka消费者会自动生成一个非空字符串。 45 properties.put("client.id", "consumer.client.id.demo"); 46 47 // 将参数设置到消费者参数中 48 KafkaConsumer consumer = new KafkaConsumer(properties); 49 50 // 消息订阅 51 consumer.subscribe(Collections.singletonList(topic)); 52 // 可以订阅多个主题 53 consumer.subscribe(Arrays.asList(topic, topic2)); 54 // 可以使用正则表达式进行订阅 55 consumer.subscribe(Pattern.compile("topic-demo*")); 56 57 // 指定订阅的分区 58 consumer.assign(Arrays.asList(new TopicPartition(topic, 0))); 59 60 while (true) { 61 // 每隔一秒监听一次 62 ConsumerRecords records = consumer.poll(Duration.ofMillis(1000)); 63 // 获取到消息信息 64 for (ConsumerRecord record : records) { 65 System.err.println(record.toString()); 66 } 67 } 68 69 } 70 71 } |
今日新闻 |
推荐新闻 |
| CopyRight 2018-2019 办公设备维修网 版权所有 豫ICP备15022753号-3 |