Kafka 生产者缓存

不建议使用：

public void produce(String message) {DmsProducer<String, String> producer = new DmsProducer<String, String>();try {producer.produce("test1",0, "key", message);} finally {producer.close();}
}

原因：
每次调用produce函数时，都会新建一个producer，每次都需要进行新建与broker连接->获取分区元数据->生产消息的过程，存在性能瓶颈
在这里插入图片描述

正确用法：

private DmsProducer producer;public void init() {this.producer = new DmsProducer();
}public void produce(String message) {if (producer == null) {this.init();}produce(message);
}public void tearDown() {if (this.producer != null) {producer.close();}
}

进阶优化：

按照上面的使用方式，如果在多线程的环境下，可能会出现一个producer单个sender线程会有性能瓶颈（单发送线程），可能导致，导致入大于出，最终缓存堆积

示例代码：

private ProducerLru cache = new ProducerLru(5);

class ProducerLru<k, v> extends LinkedHashMap<k, v> {private int capacity;public ProducerLru(int capacity) {super(capacity, 0.75f, true);this.capacity = capacity;}@Overrideprotected boolean removeEldestEntry(Map.Entry<k, v> eldest) {if (size() > capacity) {if (eldest.getValue() instanceof  DmsProducer) {DmsProducer producer = (DmsProducer) eldest.getValue();producer.close();}}return super.size() > capacity;}}public static void main(String[] args) throws Exception {Main main = new Main();String topicA = "topicA";DmsProducer producer;if (main.cache.get(topicA) != null) {producer = (DmsProducer) main.cache.get(topicA);main.cache.put(topicA, producer);} else {producer = new DmsProducer();}main.cache.put(topicA, producer);
}