不建议使用：
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;
}
@Override
protected 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);
}
上述示例中，producerLru的key可以是topic名称，如果是多个用户场景下，也可以以用户名作为key，可以根据实际业务场景进行调整
