当前位置: 首页 > news >正文

activeMq将mqtt发布订阅转成消息队列

1、activemq.xml置文件新增如下内容

2、mqttx测试发送:

主题(配置的模糊匹配,为了并发):VirtualTopic/device/sendData/12312

3、mqtt接收的结果

4、程序处理

package comimport cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;import org.apache.activemq.ActiveMQMessageAudit;
import org.apache.activemq.command.ActiveMQBytesMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;import javax.jms.Message;
import java.nio.charset.Charset;
import java.util.Date;@Component
public class MessageHandler {@JmsListener(destination = "deviceQueue.receiveDate", containerFactory = "queueListener", concurrency = "1-3")public void deviceMessage(ActiveMQBytesMessage message) {System.out.println(StrUtil.str(message.getContent().getData(), Charset.defaultCharset()));System.out.println("###################" + message + "###################");}
}

 控制台打印

 全量的:activemq.xml

<!--Licensed to the Apache Software Foundation (ASF) under one or morecontributor license agreements.  See the NOTICE file distributed withthis work for additional information regarding copyright ownership.The ASF licenses this file to You under the Apache License, Version 2.0(the "License"); you may not use this file except in compliance withthe License.  You may obtain a copy of the License athttp://www.apache.org/licenses/LICENSE-2.0Unless required by applicable law or agreed to in writing, softwaredistributed under the License is distributed on an "AS IS" BASIS,WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.See the License for the specific language governing permissions andlimitations under the License.
-->
<!-- START SNIPPET: example -->
<beansxmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsdhttp://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd"><!-- Allows us to use system properties as variables in this configuration file --><bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"><property name="locations"><value>file:${activemq.conf}/credentials.properties</value></property></bean><!--The <broker> element is used to configure the ActiveMQ broker.--><broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}"><destinationPolicy><policyMap><policyEntries><policyEntry topic=">" ><!-- The constantPendingMessageLimitStrategy is used to preventslow topic consumers to block producers and affect other consumersby limiting the number of messages that are retainedFor more information, see:http://activemq.apache.org/slow-consumer-handling.html--><pendingMessageLimitStrategy><constantPendingMessageLimitStrategy limit="1000"/></pendingMessageLimitStrategy></policyEntry></policyEntries></policyMap></destinationPolicy><!--The managementContext is used to configure how ActiveMQ is exposed inJMX. By default, ActiveMQ uses the MBean server that is started bythe JVM. For more information, see:http://activemq.apache.org/jmx.html--><managementContext><managementContext createConnector="false"/></managementContext><!--Configure message persistence for the broker. The default persistencemechanism is the KahaDB store (identified by the kahaDB tag).For more information, see:http://activemq.apache.org/persistence.html--><persistenceAdapter><kahaDB directory="${activemq.data}/kahadb"/></persistenceAdapter><!--The systemUsage controls the maximum amount of space the broker willuse before disabling caching and/or slowing down producers. For more information, see:http://activemq.apache.org/producer-flow-control.html--><systemUsage><systemUsage><memoryUsage><memoryUsage percentOfJvmHeap="70" /></memoryUsage><storeUsage><storeUsage limit="100 gb"/></storeUsage><tempUsage><tempUsage limit="50 gb"/></tempUsage></systemUsage></systemUsage><!--The transport connectors expose ActiveMQ over a given protocol toclients and other brokers. For more information, see:http://activemq.apache.org/configuring-transports.html--><transportConnectors><!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB --><transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/><transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/><transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/><transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/><transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/></transportConnectors><!-- destroy the spring context on shutdown to stop jetty --><shutdownHooks><bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" /></shutdownHooks><!-- 消息订阅 --><destinationInterceptors><virtualDestinationInterceptor><virtualDestinations><compositeTopic name="VirtualTopic.device.sendData.*"><forwardTo><queue physicalName="deviceQueue.receiveDate" /></forwardTo></compositeTopic></virtualDestinations></virtualDestinationInterceptor></destinationInterceptors><plugins><simpleAuthenticationPlugin><users><authenticationUser username="${activemq.username}" password="${activemq.password}" groups="users,admins"/><!--<authenticationUser username="user" password="password" groups="users"/><authenticationUser username="guest" password="password" groups="guests"/> --></users></simpleAuthenticationPlugin></plugins></broker><!--Enable web consoles, REST and Ajax APIs and demosThe web consoles requires by default login, you can disable this in the jetty.xml fileTake a look at ${ACTIVEMQ_HOME}/conf/jetty.xml for more details--><import resource="jetty.xml"/></beans>
<!-- END SNIPPET: example -->

http://www.lryc.cn/news/304737.html

相关文章:

  • Go语言教程
  • 分布式锁的应用场景及实现
  • 嵌入式Linux中apt、apt-get命令用法汇总
  • Unity之ShaderGraph如何实现水面波浪
  • 无线局域网(WLAN)简单概述
  • 学习数仓工具 dbt
  • 高录用快见刊【最快会后两个月左右见刊】第三届社会科学与人文艺术国际学术会议 (SSHA 2024)
  • C语言-指针初学速成
  • MQL语言实现单元测试
  • Redis信创平替之TongRDS(东方通),麒麟系统安装步骤
  • nginx服务
  • 多数pythoneer只知有列表list却不知道python也有array数组
  • 【Rust】——控制流(if-else,循环)
  • 通过platform总线驱动框架编写LED灯的驱动,编写应用程序测试
  • 费舍尔FISHER金属探测器探测仪维修F70
  • Airtest-Selenium实操小课③:下载可爱猫猫图片
  • Druid无法登录监控页面
  • 【Linux系统化学习】深入理解匿名管道(pipe)和命名管道(fifo)
  • 信息学奥赛一本通1209:分数求和
  • LabVIEW储氢材料循环寿命测试系统
  • Unity3D 框架如何搭建基于纯Lua的U框架与开发模式详解
  • Linux常见指令(2)
  • 【C++】封装
  • Maxwell安装部署
  • 说一下JVM类加载机制?
  • 解决SpringAMQP工作队列模型程序报错:WARN 48068:Failed to declare queue: simple.queue
  • mysql在服务器中的主从复制Linux下
  • QT-Day2
  • 流量分析——陇剑杯 2021【签到、jwt】
  • Java并发基础:原子类之AtomicIntegerFieldUpdater全面解析