code: private void doAsynSend(final String topicName, final String messageKey,final DefaultMessage message) {
  		// 异步发送
          this.kafkaProducer.send(new ProducerRecord<String, Object>(topicName, messageKey,message.sendBodyOnly() ? message.getBody() : message), new Callback() {
  
              @Override
              public void onCompletion(RecordMetadata metadata, Exception ex) {
                  if (ex != null) {
                  	for (ProducerEventHandler handler : eventHanlders) {
                  		try {handler.onError(topicName, message);} catch (Exception e) {}
      				}
                  	log.error("kafka_send_fail,topic="+topicName+",messageId="+messageKey,ex);
                  } else {
                  	for (ProducerEventHandler handler : eventHanlders) {
                  		try {handler.onSuccessed(topicName, metadata);} catch (Exception e) {}
      				}
                      if (log.isDebugEnabled()) {
                          log.debug("kafka_send_success,topic=" + topicName + ", messageId=" + messageKey + ", partition=" + metadata.partition() + ", offset=" + metadata.offset());
                      }
                  }
              }
          });
  	}
docs: 异步发送消息
  @param topicName
  @param messageKey
  @param message
  @param currentCount 当前执行次数