dmq代码实现

Posted 2020-12-19 water2wine

1. producer

MQProducer接口：

public interface MQProducer<K, V> {
    Future<RecordMetadata> send(ProducerRecord<K, V> var1) throws IllegalArgumentException; // 异步发送

    Future<RecordMetadata> send(ProducerRecord<K, V> var1, Callback var2) throws IllegalArgumentException; // 实现了回调函数的异步发送

    void flush(); // 

    List<PartitionInfo> partitionsFor(String var1) throws IllegalArgumentException; // 获取分区信息

    long lastOffset(TopicPartition var1) throws InterruptedException; // 获取上一次写入的偏移量

    void close(); // 关闭

    void close(long var1, TimeUnit var3); // 指定时间内关闭
}

获取需要生产的消息：

private static synchronized String getMqProducer() {
        if (null == mqProducer) {
             Properties props = new Properties();
             props.put("key.serializer", StringSerializer.class);
             props.put("value.serializer", ByteArraySerializer.class);

             mqProducer = new MQProducerImpl<>(props);
         }
        } 
        return null;
    }

发送消息：

public void sendMessage(String topic, byte[] msg) {
        AtomicBoolean result = new AtomicBoolean(true); // 使用原子类
        getMqProducer();
        ProducerRecord<String, byte[]> producerRecord = new ProducerRecord<>(topic, msg);
        mqProducer.send(producerRecord, (recordMetadata, e);
        } catch (IOException e) {
            if (null != mqProducer) {
                mqProducer.close();
            }
            result.set(false);
        }
        if (!result.get()) {
            throw new BusinessException("deal send device alarm status ");
        }
    }

2. consumer

MQConsumer接口：

public interface MQConsumer<K, V> {
    Set<TopicPartition> assignment(); // 可以订阅的topic和partition

    Set<String> subscription(); // 可以订阅的topic的String名称

    void subscribe(List<String> var1) throws IllegalArgumentException;
    void subscribe(List<String> var1, ConsumerRebalanceListener var2) throws IllegalArgumentException;
    void subscribe(Pattern var1, ConsumerRebalanceListener var2);
    void unsubscribe();

    void assign(List<TopicPartition> var1);
  
    ConsumerRecords<K, V> poll(long var1); // 拉取消息

    long position(TopicPartition var1);

    OffsetAndMetadata committed(TopicPartition var1); // 消费完后向指定partition提交偏移量和metadata

    List<PartitionInfo> partitionsFor(String var1) throws IllegalArgumentException;

    // 提交的同步和非同步方法
    void commitSync();
    void commitSync(Map<TopicPartition, OffsetAndMetadata> var1) throws IllegalArgumentException;
    void commitAsync();
    void commitAsync(OffsetCommitCallback var1) throws IllegalArgumentException;
    void commitAsync(Map<TopicPartition, OffsetAndMetadata> var1, OffsetCommitCallback var2) throws IllegalArgumentException;

    void seek(TopicPartition var1, long var2) throws IllegalArgumentException;
    void seekToBeginning(TopicPartition... var1) throws IllegalArgumentException;
    void seekToEnd(TopicPartition... var1) throws IllegalArgumentException;

    String getGroupID();

    Properties getProperties();

    void wakeup();

    void close(); // 关闭消费

    void pause(TopicPartition... var1); // 暂停消费

    void resume(TopicPartition... var1); // 
}

参数：

    public static final String GROUP_ID = "group.id"; // DMQ消息消费的GROUP_ID

    public static final String BOARD_REMOVE_GROUP_ID = "campus_alarm_service_board_remove_group_id"; // 单板移除topic的groupid

    public static final String KEY_DESERIALIZER = "key.deserializer"; // DMQ消息key的反序列化

    public static final String VALUE_DESERIALIZER = "value.deserializer"; // DMQ消息value的反序列化

    public static final String MAX_POLL_RECORDS_KEY = "max.poll.records"; // DMQ中最大pull数

    public static final String MAX_POLL_RECORDS_VALUE = "8"; // DMQ中最大pull数目

    public static final long SLEEP_TIME = 10000L;

    public static final int WAIT_FOR_POLL = 5000;

    private static final Logger LOGGER = LoggerFactory.getLogger(BoardRemoveConsumer.class); // 记录日志用

    private volatile boolean stopped; // 线程停止标志

    private final String topic; // DMQ的Topic

    private MQConsumer<String, byte[]> mqConsumer; // MQConsumer类

停止消费：

public void stop() { // 停止消费
this.stopped = true; 
mqConsumer.close();

开启消费：

public void run() { // 开启消费
        while (!stopped) {
            try {
                if (dealRun()) {
                    continue;
                }

消费的主方法：

private boolean dealRun() {
        if (null == mqConsumer) {
            // 设置消费者配置以及订阅topic
            buildPropertyAndSubscribe();
            // 如果mqConsumer为空,sleep 10s
            if (null == mqConsumer) {
                Thread.sleep(SLEEP_TIME);
                return true;
            }
        }

        // 从DMQ中获取消息
        ConsumerRecords<String, byte[]> records = mqConsumer.poll(WAIT_FOR_POLL);

        // 处理消息
        for (ConsumerRecord<String, byte[]> record : records) {
            dealRecord(record);
        }
        // 提交offset
        mqConsumer.commitSync();
        }
        return false;
    }

设置消费者配置以及订阅topic：

private void buildPropertyAndSubscribe() {
        try {
            // 设置消费者配置
            final Properties props = new Properties();
            // 每个节点都要消费DMQ消息
            props.put(GROUP_ID, BOARD_REMOVE_GROUP_ID); // 设置消费的group_id
            props.put(KEY_DESERIALIZER, StringDeserializer.class); // 设置消息的key的属性为String
            props.put(VALUE_DESERIALIZER, ByteArrayDeserializer.class); // 设置消息的value的属性为bytearray
            props.put(MAX_POLL_RECORDS_KEY, MAX_POLL_RECORDS_VALUE); // 设置最大拉取消息数

            mqConsumer = new MQConsumerImpl<String, byte[]>(props);

            // 订阅topic
            final List<String> topicList = Arrays.asList(topic);
            mqConsumer.subscribe(topicList); // 订阅
        } 
    }

处理消息：

private void dealRecord(ConsumerRecord<String, byte[]> record) {
            // 将接收到的DMQ消息转换成原对象
            NcfNotificationMsg notification = NcfNotificationMsg.parseFrom(
                record.value()); // 通过record获取原对象

            BoardRemoveBean notificationBean = null;
            notificationBean = new BoardRemoveBean(
                XmlUtil.readXmlToDocument(notification.getNotification()),
                notification.getLogicSessionId()); // 读取xml文件
            // 分发消息
            BoardRemoveImpl.getInstance().boardRemoveProcess(notificationBean);
        }
    }

线程分发消息：

void boardRemoveProcess(final BoardRemoveBean bean) {
        BoardRemoveTask task = new BoardRemoveTask(bean); // 根据topic的类型指定不同的消费任务去消费
        executor.execute(task); // 线程池执行该任务