kafka定时任务以及回调函数异步判定kafka

1、kafka定时任务：

package com.chargedot.platformservice;

import com.chargedot.platformservice.message.KafkaProducer;
import com.chargedot.platformservice.util.JsonUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import java.util.HashMap;
import java.util.Map;

/**
 * @author yanghao
 * @Description:
 * @date 2018/12/20 18:02
 */
@Component
@EnableScheduling
public class ProducerSchedule {

    @Autowired
    private KafkaProducer kafkaProducer;

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 定时任务
     */
    @Scheduled(fixedDelay = 3000)
    public void send() {
        String topic = \"C-SERVER-REQ\";
        Map<String, Object> params = new HashMap<>();
        params.put(\"OperationType\", \"SStartChargeRequest\");
        params.put(\"StartChargeSeq\", \"1354367\");
        params.put(\"UserId\", 123);
        params.put(\"StartType\", \"THRAPP\");
        params.put(\"Port\", \"\");
        String msg = JsonUtil.map2Json(params);
        System.out.println(\"-------------------------\");
        ListenableFuture future = kafkaTemplate.send(topic,\"1\" , msg);
//        Future<RecordMetadata> future = kafkaProducer.send(topic,\"1\" , msg);
//        System.out.println(\"start to send msg....\" + future);
        future.addCallback(o -> System.out.println(\"send-消息发送成功：\" + msg), throwable -> System.out.println(\"消息发送失败：\" + msg));
    }
}

2、kafka生产者判定成功失败：

ListenableFuture listenableFuture = kafkaTemplate.send(\"topic\", \"测试\");
            //发送成功后回调
            SuccessCallback successCallback = new SuccessCallback() {
                @Override
                public void onSuccess(Object result) {
                    System.out.println(\"发送成功\");
                }
            };

            //发送失败回调
            FailureCallback failureCallback = new FailureCallback() {
                @Override
                public void onFailure(Throwable ex) {
                    System.out.println(\"发送失败\");
                }
            };

 listenableFuture.addCallback(successCallback,failureCallback);