简介

去哪儿网近日宣布开源其内部广泛使用的消息中间件 QMQ 。QMQ 自 2012 年诞生以来在去哪儿网所有业务场景中广泛的应用，包括跟交易息息相关的订单场景；也包括报价搜索等高吞吐量场景。目前在公司内部日常消息 qps 在 60W 左右，生产上承载将近 4W+ 消息 topic ，消息的端到端延迟可以控制在 10ms 以内。
主要提供以下特性：
- 异步实时消息
- 延迟/定时消息
- 基于Tag的服务端过滤
- Consumer端幂等处理支持
- Consumer端filter
- 死信消息
- 结合Spring annotation使用的简单API
- 提供丰富的监控指标
- 接入OpenTracing
- 分布式事务(即将开源)
- 消息投递轨迹(即将开源)
- 历史消息的自动备份(即将开源)

demo实践

配置MetaServer

从 https://github.com/qunarcorp/qmq/releases 下载安装包，目前最新版qmq-dist-1.0.0-bin.tar.gz
解压后目录
在sql目录有个init.sql,在本地的mysql数据库中，随便建个库，运行该sql，会创建几张表
在conf中，有很多配置文件需要配置
首先是datasource.properties配置数据库地址

# 可选，MySQL驱动类

jdbc.driverClassName=com.mysql.jdbc.Driver

# 必填，MySQL数据库连接地址

jdbc.url=jdbc:mysql://<address>:<port>/<db>?<params>

# 必填，MySQL数据库用户名

jdbc.username=

# 必填，MySQL数据库密码

jdbc.password=

# 可选，连接池最大连接数

pool.size.max=10

metaserver.properties

#可选，提供http服务，用于meta server的服务发现

meta.server.discover.port=8080

#可选，以tcp的方式监听，供client和server访问

meta.server.port=20880

# 可选，内部数据缓存刷新间隔

refresh.period.seconds=5

# 可选，动态生效，broker心跳超时时间

heartbeat.timeout.ms=30000

# 可选，动态生效，每个主题至少分配多少broker group

min.group.num=2

valid-api-tokens.properties（一行一条白名单，在加入集群服务时配置需要用到，设置=前面的参数即可）

123=default-pass

client_log_switch.properties

# 是否输出所有主题的详细请求信息

default=false

# 可以控制单个主题是否输出详细请求信息

<subject a>=true

<subject b>=false

使用bin目录的metaserver.sh(windows平台上请使用metaserver.cmd)启动

配置Broker

# 必填，metaserver地址，即你第一步安装的meta server的ip地址，注意这里的地址的端口是meta.server.discover.port指定的端口，默认是8080

meta.server.endpoint=http://<metaserver address>/meta/address

# 可选，broker服务端口

broker.port=20881

# 可选，同步数据端口

sync.port=20882

# 可选，动态生效，从机同步请求超时时间

slave.sync.timeout=3000

# 必填，数据存放目录

store.root=/data

# 可选，动态生效，主是否等待从写入完成再返回写入结果

wait.slave.wrote=false

# 可选，动态生效，重试消息延迟派发时间

message.retry.delay.seconds=5

# 可选，动态生效，messagelog过期时间

messagelog.retention.hours=72

# 可选，动态生效，consumerlog过期时间

consumerlog.retention.hours=72

# 可选，动态生效，pulllog过期时间

pulllog.retention.hours

# 可选，动态生效，数据文件过期检查周期

log.retention.check.interval.seconds

# 可选，动态生效，是否删除过期数据

log.expired.delete.enable=true

# 可选，动态生效，checkpoint文件保留数量

checkpoint.retain.count=5

# 可选，动态生效，action checkpoint强制写入周期，单位为日志数量

action.checkpoint.interval=100000

# 可选，动态生效，message checkpoint强制写入周期，单位为日志数量

message.checkpoint.interval=100000

# 可选，动态生效，重试消息写入QPS限制

put_need_retry_message.limiter=50

# 可选，动态生效，从机一次最多拉取多少数据

sync.batch.size=100000

# 可选，动态生效，从机同步数据超时时间

message.sync.timeout.ms=10

运行,liunx使用tools.sh

tools.cmd AddBroker --metaserver=localhost:8080 --token=123 --brokerGroup=group1 --role=0 --hostname=localhost --ip=127.0.0.1 --servePort=20881 --syncPort=20882

metaserver address指的是ip:port,port默认是8080
token即metaserver的配置valid-api-tokens.properties里任何一项
brokerGroup 这一组的名字，每一组分为主从
role 角色 0 - master, 1 - slave, 5 - delay master, 6 - delay slave
hostname broker的主机名
ip broker的ip地址
servePort broker接收消息的端口
syncPort 主从同步端口

配置Delay Server

# 必填，metaserver地址，即你第一步安装的meta server的ip地址，注意这里的地址的端口是meta.server.discover.port指定的端口，默认是8080

meta.server.endpoint=http://<metaserver address>/meta/address

# 可选，broker服务端口

broker.port=20881

# 可选，同步数据端口

sync.port=20882

# 可选，动态生效，从机同步请求超时时间

slave.sync.timeout=3000

# 必填，数据存放目录

store.root=/data

# 可选，动态生效，主是否等待从写入完成再返回写入结果

wait.slave.wrote=true

# 可选，动态生效，从机一次最多拉取多少数据

sync.batch.size=100000

# 可选，动态生效，从机同步dispatch数据超时时间

dispatch.sync.timeout.ms=10

# 可选，动态生效，从机同步message数据超时时间

message.sync.timeout.ms=10

# 可选，动态生效，是否删除过期数据

log.expired.delete.enable=true

# 可选，动态生效，数据文件过期检查周期

log.retention.check.interval.seconds=60

# 可选，动态生效，dispatch文件过期时间

dispatch.log.keep.hour=72

# 可选，动态生效，messagelog过期时间

messagelog.retention.hours=72

运行

tools.cmd AddBroker --metaserver=localhost:8080 --token=123 --brokerGroup=group1 --role=5 --hostname=localhost --ip=127.0.0.1 --servePort=20881 --syncPort=20882

运行失败~

broker.cmd 启动并报错后，metaserver也报错
broker报错

[2018-12-17 16:33:16 ERROR qunar.tc.qmq.meta.BrokerRegisterService] Send acquire meta message to meta server failed

java.lang.IndexOutOfBoundsException: null

        at io.netty.buffer.EmptyByteBuf.readShort(EmptyByteBuf.java:450)

        at qunar.tc.qmq.utils.PayloadHolderUtils.readString(PayloadHolderUtils.java:37)

        at qunar.tc.qmq.meta.BrokerAcquireMetaResponseSerializer.deSerialize(BrokerAcquireMetaResponseSerializer.java:31)

        at qunar.tc.qmq.meta.BrokerRegisterService.acquireMeta(BrokerRegisterService.java:76)

        at qunar.tc.qmq.meta.BrokerRegisterService.start(BrokerRegisterService.java:68)

        at qunar.tc.qmq.startup.ServerWrapper.register(ServerWrapper.java:103)

        at qunar.tc.qmq.startup.ServerWrapper.start(ServerWrapper.java:86)

        at qunar.tc.qmq.container.Bootstrap.main(Bootstrap.java:29)

Exception in thread "main" java.lang.RuntimeException: java.lang.IndexOutOfBoundsException

        at qunar.tc.qmq.meta.BrokerRegisterService.acquireMeta(BrokerRegisterService.java:80)

        at qunar.tc.qmq.meta.BrokerRegisterService.start(BrokerRegisterService.java:68)

        at qunar.tc.qmq.startup.ServerWrapper.register(ServerWrapper.java:103)

        at qunar.tc.qmq.startup.ServerWrapper.start(ServerWrapper.java:86)

        at qunar.tc.qmq.container.Bootstrap.main(Bootstrap.java:29)

Caused by: java.lang.IndexOutOfBoundsException

        at io.netty.buffer.EmptyByteBuf.readShort(EmptyByteBuf.java:450)

        at qunar.tc.qmq.utils.PayloadHolderUtils.readString(PayloadHolderUtils.java:37)

        at qunar.tc.qmq.meta.BrokerAcquireMetaResponseSerializer.deSerialize(BrokerAcquireMetaResponseSerializer.java:31)

        at qunar.tc.qmq.meta.BrokerRegisterService.acquireMeta(BrokerRegisterService.java:76)

metaServer报错

[2018-12-17 16:33:16 ERROR qunar.tc.qmq.netty.NettyRequestExecutor] doExecute request exception, channel:[id: 0x88b9cfb5, L:/172.16.100.141:20880 - R:/172.16.100.141:59614], cmd:Datagram{header=RemotingHeader{magicCode=-557774114, code=40, version=8, opaque=0, flag=0, requestCode=-1}}

java.lang.RuntimeException: cannot find broker meta for DESKTOP-BCBMBTF:20881

        at qunar.tc.qmq.meta.processor.BrokerAcquireMetaProcessor.lambda$createResponse$1(BrokerAcquireMetaProcessor.java:68)

        at java.util.Optional.orElseThrow(Optional.java:290)

        at qunar.tc.qmq.meta.processor.BrokerAcquireMetaProcessor.createResponse(BrokerAcquireMetaProcessor.java:68)

        at qunar.tc.qmq.meta.processor.BrokerAcquireMetaProcessor.processRequest(BrokerAcquireMetaProcessor.java:58)

        at qunar.tc.qmq.netty.NettyRequestExecutor.doExecute(NettyRequestExecutor.java:86)

        at qunar.tc.qmq.netty.NettyRequestExecutor.executeWithMonitor(NettyRequestExecutor.java:70)

        at qunar.tc.qmq.netty.NettyRequestExecutor.execute(NettyRequestExecutor.java:56)

        at qunar.tc.qmq.netty.NettyServerHandler.processRequestCommand(NettyServerHandler.java:79)

        at qunar.tc.qmq.netty.NettyServerHandler.processMessageReceived(NettyServerHandler.java:57)

        at qunar.tc.qmq.netty.NettyServerHandler.channelRead0(NettyServerHandler.java:50)

        at qunar.tc.qmq.netty.NettyServerHandler.channelRead0(NettyServerHandler.java:37)

        at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)

        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:292)

        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:278)

        at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:277)

        at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:264)

        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:292)

        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:278)

        at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)

        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:292)

        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:278)

        at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:962)

        at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)

        at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:528)

        at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:485)

        at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:399)

        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:371)

        at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:112)

        at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)

        at java.lang.Thread.run(Thread.java:748)

总结

qmq刚开源出来，很多东西有待实践，我碰到的问题目前网上很难搜到答案
若有大神运行成功官方demo的，还请写篇博客，必拜访推荐，请在评论区留下地址

官方地址

https://github.com/qunarcorp/qmq

秒客网

QMQ去哪儿网-mq中间件（启动失败）

简介

demo实践

配置MetaServer

配置Broker

配置Delay Server

运行失败~

总结

相关文章