微信公众号搜"智元新知"关注
微信扫一扫可直接关注哦!

qmq-spring-boot-starter 启用消费者模式配置消费监听器

程序名称:qmq-spring-boot-starter 启用消费者模式配置消费监听器

授权协议: Apache

操作系统: 跨平台

开发语言: Java

qmq-spring-boot-starter 启用消费者模式配置消费监听器 介绍

使用方式

引入 Maven 依赖(已上传到中央仓库)

<dependency>
    <groupId>xin.wjtree.qmq</groupId>
    <artifactId>qmq-spring-boot-starter</artifactId>
    <version>1.0.0</version>
</dependency>

添加 Spring Boot 配置(YML)

spring:
  application:
    name: qmq-demo
  qmq:
    # 应用标识 appcode,必填
    app-code: qmq-demo
    # 服务器地址 Metaserver,必填
    Meta-server: http://127.0.0.1:8080/Meta/address

    # 生产者配置,发送消息的线程池的设置,选填
    producer:
      # 发送线程数,认 3
      send-threads: 3
      # 认每次发送时最大批量大小,认 30
      send-batch: 30
      # 如果消息发送失败,重试次数认 10
      send-try-count: 10
      # 异步发送队列大小,认 10000
      max-queue-size: 10000

    # 使用 QmqTemplate 发送消息的主题认值 default_subject
    template:
      default-subject: default_subject

    # 消费者配置,消费消息的线程池的设置,选填
    consumer:
      # 线程名称前缀,认 qmq-process
      thread-name-prefix: qmq-process
      # 线程池大小,认 2
      core-pool-size: 2
      # 最大线程池大小,认 2
      max-pool-size: 2
      # 线程池队列大小,认 1000
      queue-capacity: 1000

    # 消息主题和分组配置,选填
    # 使用 QmqConsumer 注解时,可使用 SpEL 表达式引入以下主题和分组
    subject:
      sub1: sub1
      sub2: sub2
      sub3: sub3
      # more subject ...
    group:
      group1: group1
      group2: group2
      group3: group3
      # more group ...

logging:
  level:
    # 设置 qmq-spring-boot-starter 的日志级别
    xin.wjtree.qmq: trace

server:
  port: 8989

发送消息

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBoottest;
import org.springframework.test.context.junit4.springrunner;
import qunar.tc.qmq.Message;
import qunar.tc.qmq.MessageSendStateListener;
import xin.wjtree.qmq.QmqTemplate;
import xin.wjtree.qmq.autoconfigure.QmqProperties;
import xin.wjtree.qmq.constant.QmqTimeUnit;
import xin.wjtree.qmq.internal.QmqAlias;
import xin.wjtree.qmq.internal.QmqIgnore;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.CountDownLatch;

@RunWith(springrunner.class)
@SpringBoottest
public class QmqTest {
    @Resource
    private QmqTemplate template;
    @Resource
    private QmqProperties properties;

    /**
     * 发送即时消息
     * @throws InterruptedException
     */
    @Test
    public void sendImmediate() throws InterruptedException {
        // 计数器,执行1次结束
        CountDownLatch latch = new CountDownLatch(1);

        // 一般使用 template.send(properties.getSubject().get("sub1"), getUser()) 即可
        template.withSendStateListener(new MessageSendStateListener() {
            @Override
            public void onSuccess(Message m) {
                latch.countDown();
            }

            @Override
            public void onFailed(Message m) {
                latch.countDown();
            }
        }).send(properties.getSubject().get("sub1"), getUser());

        // 计数器减1
        latch.await();
    }

    /**
     * 发送延时消息
     * @throws InterruptedException
     */
    @Test
    public void sendDelay() throws InterruptedException {
        // 计数器,执行1次结束
        CountDownLatch latch = new CountDownLatch(1);

        // 延时 10 秒发送消息
        // 一般使用 template.sendDelay(properties.getSubject().get("sub1"), getUser(), QmqTimeUnit.TEN_SECONDS) 即可
        template.withSendStateListener(new MessageSendStateListener() {
            @Override
            public void onSuccess(Message m) {
                latch.countDown();
            }

            @Override
            public void onFailed(Message m) {
                latch.countDown();
            }
        }).sendDelay(properties.getSubject().get("sub1"), getUser(), QmqTimeUnit.TEN_SECONDS);

        // 计数器减1
        latch.await();
    }

    /**
     * 发送定时消息
     * @throws InterruptedException
     */
    @Test
    public void sendSchedule() throws InterruptedException, ParseException {
        // 计数器,执行1次结束
        CountDownLatch latch = new CountDownLatch(1);

        // 定时发送的日期时间
        Date date = new SimpleDateFormat("yyyy-MM-dd HH

qmq-spring-boot-starter 启用消费者模式配置消费监听器 官网

https://gitee.com/wjtree/qmq-spring-boot-starter

版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 [email protected] 举报,一经查实,本站将立刻删除。

相关推荐