上程式碼, 包括Pulsar的引數類, Pulsar Client, Producer和Consumer

================Pulsar引數類=====================
@Data
@RefreshScope
@Component
@ConfigurationProperties(prefix = "tdmq.pulsar")
public class PulsarProperties {

    /**
     * 接入地址
     */
    private String serviceurl;

    /**
     * 名稱空間tdc
     */
    private String tdcNamespace;

    /**
     * 角色tdc的token
     */
    private String tdcToken;

    /**
     * 叢集name
     */
    private String cluster;

/**
 * topicMap
 */
private Map<String, String> topicMap;

/**
 * 訂閱
 */
private Map<String, String> subMap;

/**
 * 開關 on:Consumer可用 ||||| off:Consumer斷路
 */
private String onOff;

}
==================PulsarClient=======================

@Slf4j
@Configuration
@EnableConfigurationProperties(PulsarProperties.class)
public class PulsarConfig {

    @Autowired
    PulsarProperties pulsarProperties;

    @RefreshScope
    @Bean
    public PulsarClient getPulsarClient() {

        try {
            return PulsarClient.builder()
                    .authentication(AuthenticationFactory.token(pulsarProperties.getTdcToken()))
                    .serviceUrl(pulsarProperties.getServiceurl())
                    .build();
        } catch (PulsarClientException e) {
            log.error("初始化Pulsar Client失敗", e);
        }

        throw new RuntimeException("初始化Pulsar Client失敗");
    }

}
===========Producer&Consumer&傳送訊息的工具類=================

@Slf4j
@Component
public class PulsarUtils {

    @Autowired
    PulsarProperties pulsarProperties;

    @Autowired
    PulsarClient client;

    @Autowired
    AuditCommentResultListener<String> auditCommentResultListener;

    @Autowired
    AuditReplyResultListener<String> auditReplyResultListener;

    @Autowired
    AuditResourceResultListener<String> auditResourceResultListener;

    /**
     * 建立一個生產者
     *
     * @param topic topic name
     * @return Producer生產者
     */
    public Producer<byte[]> createProducer(String topic) {

        try {
            return client.newProducer()
                    .topic(pulsarProperties.getCluster() + "/" + pulsarProperties.getTdcNamespace() + "/" + topic)
                    .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
                    .sendTimeout(10, TimeUnit.SECONDS)
                    .blockIfQueueFull(true)
                    .create();
        } catch (PulsarClientException e) {
            log.error("初始化Pulsar Producer失敗", e);
        }

        throw new RuntimeException("初始化Pulsar Producer失敗");
    }

    /**
     * 建立一個消費者
     *
     * @param topic           topic name
     * @param subscription    sub name
     * @param messageListener MessageListener的自定義實現類
     * @return Consumer消費者
     */
    public Consumer createConsumer(String topic, String subscription,
                                   MessageListener messageListener) {
        try {
            return client.newConsumer()
                    .topic(pulsarProperties.getCluster() + "/" + pulsarProperties.getTdcNamespace() + "/" + topic)
                    .subscriptionName(subscription)
                    .ackTimeout(10, TimeUnit.SECONDS)
                    .subscriptionType(SubscriptionType.Shared)
                    .messageListener(messageListener)
                    .subscribe()
                    ;
        } catch (PulsarClientException e) {
            log.error("初始化Pulsar Consumer失敗", e);
        }

        throw new RuntimeException("初始化Pulsar Consumer失敗");
    }

    /**
     * 非同步send一條msg
     *
     * @param message 訊息體
     */
    public void sendMessage(String message, Producer<byte[]> producer) {
        producer.sendAsync(message.getBytes()).thenAccept(msgId -> {
            log.info("訊息傳送成功, MessageID為{}", msgId);
        });
    }

    /**
     * 同步傳送一條msg
     *
     * @param message  訊息體
     * @param producer 生產者例項
     */
    public void sendOnce(String message, Producer<byte[]> producer) throws PulsarClientException {
        MessageId send = producer.send(message.getBytes());
        log.info("訊息成功傳送, MessageId {},message {}", send, message);
    }

    //-----------consumer-----------
    @RefreshScope
    @Bean(name = "audit-resource-result-topic")
    public Consumer getAuditResourceResultTopicConsumer() {
        return this.createConsumer(pulsarProperties.getTopicMap().get("audit-resource-result-topic"),
                pulsarProperties.getSubMap().get("resource-sub-audit-resource-result"),
                auditResourceResultListener);
    }

    //-----------producer-----------
    @RefreshScope
    @Bean(name = "resource-publish-topic")
    public Producer<byte[]> getResourcePublishTopicProducer() {
        return this.createProducer(pulsarProperties.getTopicMap().get("resource-publish-topic"));
    }
}
=====================AbstractListener===============================

@Slf4j
@Component
public abstract class AbstractListener<String> implements MessageListener<String> {

    @Autowired
    PulsarProperties pulsarProperties;

    @Override
    public void received(Consumer<String> consumer, Message<String> message) {

    }

    /**
     * 判斷開關
     *
     * @return is equals off
     */
    public boolean judgeIsOff() {
        return pulsarProperties.getOnOff().equals("off");
    }
}

=================Listener自定義實現類====================

@Slf4j
@Component
public class AuditCommentResultListener<String> extends AbstractListener<String> {

    @Autowired
    CommentService commentService;

    @Override
    public void received(Consumer consumer, Message msg) {
        try {
            java.lang.String data = new java.lang.String(msg.getData());
            log.info("接受到訊息, MessageId {} data {}", msg.getMessageId(), data);
            // 新增開關
            if (super.judgeIsOff()) {
                consumer.negativeAcknowledge(msg);
                log.error("當前開關為off 拒絕消費訊息, MessageId {} data {}", msg.getMessageId(), data);
            }
            // 處理業務邏輯

            consumer.acknowledge(msg);
        } catch (Exception e) {
            consumer.negativeAcknowledge(msg);
            log.error("拒絕消費訊息, MessageId {} data {}", msg.getMessageId(), new java.lang.String(msg.getData()), e);
        }
    }
}

=========================================================================