消息中间件JMQ客户端使用

编程语言 2019-03-02 13:32:36 阅读次数: 0

1. JMQ客户端(目前版本2.1.5-SNAPSHOT)特性

  • 不依赖于第三方组件，只和Broker通信

  • 内置管理和性能采集协议

  • 较合理的超时时间设置

  • 批量发送和消费，消费者为拉模式

  • 支持两阶段事务

  • 支持业务ID顺序消费

  • 支持机房部署，就近发送和消费

  • 默认数据压缩，更快的性能

  • 支持流量分流控制

2. Maven依赖

2.1.Spring支持：

< dependency >
    < groupId >com.jd.jmq</ groupId >
    < artifactId >jmq-client-spring</ artifactId >
    < version >2.1.5-SNAPSHOT</ version >
</ dependency >
 
< dependency >
     < groupId >com.jd.ump</ groupId >
     < artifactId >profiler</ artifactId >
     < version >3.2.0</ version > <!-- 必须3.2.0及以上！！！ -->
</ dependency >

schema配置(2.1.5-SNAPSHOT版本)

< beans ... xmlns:jmq = "http://code.jd.com/schema/jmq"
         xsi:schemaLocation="...
         http://code.jd.com/schema/jmq
         http://code.jd.com/schema/jmq/jmq-1.0.xsd">


     schema配置(2.1.0以下版本，包括2.1.0)

< beans ... xmlns:jmq = "http://code.jd.com/schema/jmq"
         xsi:schemaLocation="...
         http://code.jd.com/schema/jmq
         http://code.jd.com/schema/jmq/jmq-1.0.xsd">

 schema配置(2.1.1以上版本，包括2.1.1，不包括 2.1.5-SNAPSHOT)

< beans ... xmlns:jmq = "http://code.jd.com/schema/jmq"
         xsi:schemaLocation="...
         http://code.jd.com/schema/jmq
         http://code.jd.com/schema/jmq/jmq-1.1.xsd">

2.2 API调用：

< dependency >
    < groupId >com.jd.jmq</ groupId >
    < artifactId >jmq-client-core</ artifactId >
    < version >2.1.5-SNAPSHOT</ version >
</ dependency >
< dependency >
    < groupId >com.jd.jmq</ groupId >
    < artifactId >jmq-client-ump</ artifactId >
    < version >2.1.5-SNAPSHOT</ version >
</ dependency >
 
< dependency >
     < groupId >com.jd.ump</ groupId >
     < artifactId >profiler</ artifactId >
     < version >3.2.0</ version >
</ dependency >

3.配置项

    重要说明，在配置中或者应用代码中使用到的topic都不需要加任何前缀，填写申请时的主题代码即可

3.1 连接通道配置：

  • TransportConfig 常用属性说明(此类型对应spring配置文件中的 <jmq:transport .../>标签)
属性
说明
默认值
address 目录服务地址 
JMQ接入地址
user  用户  查看链接用户名和密码（TOKEN信息）
password  密码  查看链接用户名和密码（TOKEN信息）

app

 接入的应用（系统）代码，此属性必须填写，和user完全一样 nullAPP不同的TOPIC不能共用transport
sendTimeout 发送超时（毫秒） 5000测试环境请适度改大
epoll 是否使用epoll模式，会根据系统环境自动设置 true
  • 线上环境配置样例：
< jmq:transport address = "jmq-cluster.jd.local:80" user = "${jmq.user}" password = "${jmq.password}" app = "${jmq.producer.app}" />
  • 测试环境配置样例：
< jmq:transport address = "jmq-testcluster.jd.local:50088" password = "${jmq.password}" app = "${jmq.producer.app}" />

3.2 发送端配置

  • MessageProducer属性说明(此类型对应spring配置文件中的 <jmq:producer  id="producer".../>标签)
属性
说明
默认值
transport 连接通道 jmq.transport
retryTimes 发送失败后的重试次数 2
  • 配置样例（请参考demo程序的spring-producer.xml）
< jmq:producer id = "producer" retryTimes = "2" transport = "jmq.transport" />

注：在jmq管理平台， 暂停消费只能用于紧急情况，如果长时间暂停会导致消息积压，服务端消息文件无法删除，磁盘会被写满


  • 发送代码样例(请参考 demo里的ProducerTest.java)
//获取实例
Producer producer= (Producer) context.getBean( "producer" );
 
//普通发送方式
Message message = new Message( "topic" , "消息内容" , "业务ID" );   //业务ID不要超过16个字节，不然归档会有问题， 超过100个字节重试会有问题
producer.send(message, timeout);
 
//批量发送方式
List<Message> messages = new ArrayList<Message>();
for ( int j = 0 ; j < n; j++) {
     Message message = new Message( "topic" , "消息内容" , "业务ID" );
     messages.add(message);
}
 
producer.send(messages);
  • 事物发送样例
transaction
/**
     * 事务的单条发送
     */
    protected void sendTransaction( int i) throws JMQException {
        Message message = new Message(topic, topic + "_test_" + i, topic + "_rid_" + i);
        Boolean ret = producer.send(message, new LocalTransaction<Boolean>() {
            @Override
            public Boolean execute() throws Exception {
                System.out.println( "execute local transaction ok!" );
                return true ;
            }
 
            @Override
            public int getTimeout() {
                return 50000 ;
            }
        });
        System.out.println( "local transaction ret value:" + ret);
    }


3.3 消费端配置（分为监听器模式和拉取模式）：

       重要说明：JMQ不再区分正常消息和重试消息，请不要再根据消息属性来判断是否是重试消息。

  • 消费容器属性说明
属性
说明
默认值
transport 连接工厂 jmq.transport
autoStart 自动启动消费 true，》=1.2.25以上版本支持，高级特性谨慎设置
pullTimeout 长轮询超时时间（一般不需要配置） 10000ms  pullTimeout必须大于longPull时间，否则客户端会报"请求超时" 
longPull
 长轮询等待时间（一般不需要配置） 5000ms 
maxConcurrent

连接单组broker最大并发数,客户端实际线程数为单组运行的线程数*broker数（一般不需要配置，默认等于对列数）
  topic索引分区数，管理端配置 》=1.3.1以上版本支持 
minConcurrent
 
连接单组broker最小并发数（一般不需要配置）
 0 自动调节》=1.3.1以上版本支持
3.3.1 监听器模式：


  • 消息监听器属性说明
属性
说明
默认值
topic 消费者topic串，必须注入 null
listener 消息监听器是MessageListener接口的实现类所创建的对象，用于处理收到的消息
抛出异常将回滚重试消息，否则成功消费并删除消息，所以如果认为消息处理失败还需重试请抛出异常		 null
  • 监听器模式配置样例（请参考demo里的spring-consumer.xml）
< jmq:consumer id = "consumer" transport = "jmq.transport" >
     < jmq:listener topic = "${jmq.consumer.topic}" listener = "messageListener" />
</ jmq:consumer >
 
< bean id = "messageListener" class = "com.jd.jmq.demo.DefaultMessageListener" />
3.3.2拉取模式：

拉取模式指用户在代码里主动调用pull方法，不需要在配置文件里面再配置<jmq:listener />，拉取的速度由用户控制，调用一次拉取一次消息进行消费，用户自己启用多线程控制并行度。

method: pull(String topic,MessageListener listener)

topic:指消费的主题名

listener:是一个回调对象，当pull拉取到消息后会主动调用listener.onMessage()，

与监听模式的区别是:监听模式由JMQ客户端守护线程去不停的拉取消息进行消费，拉取模式由用户控制拉取的频率，不主动调用就不会消费消息。但是都不需要主动对消息进行确认


  • 代码样例（请参考demo里的ApiConsumer.java, 路径  src\test\java\com\jd\jmq\demo ）
messageConsumer.start();
for (;;){
     //手动拉取消息
     messageConsumer.pull(topic,messageListener);
}
  • 消息处理器代码样例(请参考demo里的DefaultMessageListener.java)
/**
  * 消费方法。注意: 消费不成功请抛出异常，JMQ会自动重试
  *
  * @param messages
  * @throws Exception
  */
@Override
public void onMessage(List<Message> messages) throws Exception {
     if (messages == null || messages.isEmpty()) {
         return ;
     }
 
     for (Message message : messages) {
         logger.info(String.format( "收到一条消息,消息主题（队列名）：%s,内容是：%s" , message.getTopic(), message.getText()));
     }
}

1. JMQ客户端(目前版本2.1.5-SNAPSHOT)特性

  • 不依赖于第三方组件，只和Broker通信

  • 内置管理和性能采集协议

  • 较合理的超时时间设置

  • 批量发送和消费，消费者为拉模式

  • 支持两阶段事务

  • 支持业务ID顺序消费

  • 支持机房部署，就近发送和消费

  • 默认数据压缩，更快的性能

  • 支持流量分流控制

2. Maven依赖

2.1.Spring支持：

< dependency >
    < groupId >com.jd.jmq</ groupId >
    < artifactId >jmq-client-spring</ artifactId >
    < version >2.1.5-SNAPSHOT</ version >
</ dependency >
 
< dependency >
     < groupId >com.jd.ump</ groupId >
     < artifactId >profiler</ artifactId >
     < version >3.2.0</ version > <!-- 必须3.2.0及以上！！！ -->
</ dependency >

schema配置(2.1.5-SNAPSHOT版本)

< beans ... xmlns:jmq = "http://code.jd.com/schema/jmq"
         xsi:schemaLocation="...
         http://code.jd.com/schema/jmq
         http://code.jd.com/schema/jmq/jmq-1.0.xsd">


     schema配置(2.1.0以下版本，包括2.1.0)

< beans ... xmlns:jmq = "http://code.jd.com/schema/jmq"
         xsi:schemaLocation="...
         http://code.jd.com/schema/jmq
         http://code.jd.com/schema/jmq/jmq-1.0.xsd">

 schema配置(2.1.1以上版本，包括2.1.1，不包括 2.1.5-SNAPSHOT)

< beans ... xmlns:jmq = "http://code.jd.com/schema/jmq"
         xsi:schemaLocation="...
         http://code.jd.com/schema/jmq
         http://code.jd.com/schema/jmq/jmq-1.1.xsd">

2.2 API调用：

< dependency >
    < groupId >com.jd.jmq</ groupId >
    < artifactId >jmq-client-core</ artifactId >
    < version >2.1.5-SNAPSHOT</ version >
</ dependency >
< dependency >
    < groupId >com.jd.jmq</ groupId >
    < artifactId >jmq-client-ump</ artifactId >
    < version >2.1.5-SNAPSHOT</ version >
</ dependency >
 
< dependency >
     < groupId >com.jd.ump</ groupId >
     < artifactId >profiler</ artifactId >
     < version >3.2.0</ version >
</ dependency >

3.配置项

    重要说明，在配置中或者应用代码中使用到的topic都不需要加任何前缀，填写申请时的主题代码即可

3.1 连接通道配置：

  • TransportConfig 常用属性说明(此类型对应spring配置文件中的 <jmq:transport .../>标签)
属性
说明
默认值
address 目录服务地址 
JMQ接入地址
user  用户  查看链接用户名和密码（TOKEN信息）
password  密码  查看链接用户名和密码（TOKEN信息）

app

 接入的应用（系统）代码，此属性必须填写，和user完全一样 nullAPP不同的TOPIC不能共用transport
sendTimeout 发送超时（毫秒） 5000测试环境请适度改大
epoll 是否使用epoll模式，会根据系统环境自动设置 true
  • 线上环境配置样例：
< jmq:transport address = "jmq-cluster.jd.local:80" user = "${jmq.user}" password = "${jmq.password}" app = "${jmq.producer.app}" />
  • 测试环境配置样例：
< jmq:transport address = "jmq-testcluster.jd.local:50088" password = "${jmq.password}" app = "${jmq.producer.app}" />

3.2 发送端配置

  • MessageProducer属性说明(此类型对应spring配置文件中的 <jmq:producer  id="producer".../>标签)
属性
说明
默认值
transport 连接通道 jmq.transport
retryTimes 发送失败后的重试次数 2
  • 配置样例（请参考demo程序的spring-producer.xml）
< jmq:producer id = "producer" retryTimes = "2" transport = "jmq.transport" />

注：在jmq管理平台， 暂停消费只能用于紧急情况，如果长时间暂停会导致消息积压，服务端消息文件无法删除，磁盘会被写满


  • 发送代码样例(请参考 demo里的ProducerTest.java)
//获取实例
Producer producer= (Producer) context.getBean( "producer" );
 
//普通发送方式
Message message = new Message( "topic" , "消息内容" , "业务ID" );   //业务ID不要超过16个字节，不然归档会有问题， 超过100个字节重试会有问题
producer.send(message, timeout);
 
//批量发送方式
List<Message> messages = new ArrayList<Message>();
for ( int j = 0 ; j < n; j++) {
     Message message = new Message( "topic" , "消息内容" , "业务ID" );
     messages.add(message);
}
 
producer.send(messages);
  • 事物发送样例
transaction
/**
     * 事务的单条发送
     */
    protected void sendTransaction( int i) throws JMQException {
        Message message = new Message(topic, topic + "_test_" + i, topic + "_rid_" + i);
        Boolean ret = producer.send(message, new LocalTransaction<Boolean>() {
            @Override
            public Boolean execute() throws Exception {
                System.out.println( "execute local transaction ok!" );
                return true ;
            }
 
            @Override
            public int getTimeout() {
                return 50000 ;
            }
        });
        System.out.println( "local transaction ret value:" + ret);
    }


3.3 消费端配置（分为监听器模式和拉取模式）：

       重要说明：JMQ不再区分正常消息和重试消息，请不要再根据消息属性来判断是否是重试消息。

  • 消费容器属性说明
属性
说明
默认值
transport 连接工厂 jmq.transport
autoStart 自动启动消费 true，》=1.2.25以上版本支持，高级特性谨慎设置
pullTimeout 长轮询超时时间（一般不需要配置） 10000ms  pullTimeout必须大于longPull时间，否则客户端会报"请求超时" 
longPull
 长轮询等待时间（一般不需要配置） 5000ms 
maxConcurrent

连接单组broker最大并发数,客户端实际线程数为单组运行的线程数*broker数（一般不需要配置，默认等于对列数）
  topic索引分区数，管理端配置 》=1.3.1以上版本支持 
minConcurrent
 
连接单组broker最小并发数（一般不需要配置）
 0 自动调节》=1.3.1以上版本支持
3.3.1 监听器模式：


  • 消息监听器属性说明
属性
说明
默认值
topic 消费者topic串，必须注入 null
listener 消息监听器是MessageListener接口的实现类所创建的对象，用于处理收到的消息
抛出异常将回滚重试消息，否则成功消费并删除消息，所以如果认为消息处理失败还需重试请抛出异常		 null
  • 监听器模式配置样例（请参考demo里的spring-consumer.xml）
< jmq:consumer id = "consumer" transport = "jmq.transport" >
     < jmq:listener topic = "${jmq.consumer.topic}" listener = "messageListener" />
</ jmq:consumer >
 
< bean id = "messageListener" class = "com.jd.jmq.demo.DefaultMessageListener" />
3.3.2拉取模式：

拉取模式指用户在代码里主动调用pull方法，不需要在配置文件里面再配置<jmq:listener />，拉取的速度由用户控制，调用一次拉取一次消息进行消费，用户自己启用多线程控制并行度。

method: pull(String topic,MessageListener listener)

topic:指消费的主题名

listener:是一个回调对象，当pull拉取到消息后会主动调用listener.onMessage()，

与监听模式的区别是:监听模式由JMQ客户端守护线程去不停的拉取消息进行消费，拉取模式由用户控制拉取的频率，不主动调用就不会消费消息。但是都不需要主动对消息进行确认


  • 代码样例（请参考demo里的ApiConsumer.java, 路径  src\test\java\com\jd\jmq\demo ）
messageConsumer.start();
for (;;){
     //手动拉取消息
     messageConsumer.pull(topic,messageListener);
}
  • 消息处理器代码样例(请参考demo里的DefaultMessageListener.java)
/**
  * 消费方法。注意: 消费不成功请抛出异常，JMQ会自动重试
  *
  * @param messages
  * @throws Exception
  */
@Override
public void onMessage(List<Message> messages) throws Exception {
     if (messages == null || messages.isEmpty()) {
         return ;
     }
 
     for (Message message : messages) {
         logger.info(String.format( "收到一条消息,消息主题（队列名）：%s,内容是：%s" , message.getTopic(), message.getText()));
     }
}

猜你喜欢

转载自blog.csdn.net/weixin_42537413/article/details/88074028
今日推荐
周排行
成为C++高手之宏与枚举
在CAD二次开发中使用进度条
Js插件ECharts，HighCharts学习网址整理
Celery提交任务出错(on windows.)
cephfs内核客户端性能追踪
thinkphp中PHPExcel用法
EntityFramework动态组合多排序字段
汇编语言（八）实验9 根据材料编程
安装ubuntu后必须做的事情（对我而言）
JS函数式编程
每日归档
更多
2024-10-22(0)
2024-10-21(0)
2024-10-20(0)
2024-10-19(0)
2024-10-18(0)
2024-10-17(0)
2024-10-16(0)
2024-10-15(0)
2024-10-14(0)
2024-10-13(0)

代码天地 © 2018 codetd.com

境外服务器   最新电视剧2022