折800 网站模板,2016网站备案,wordpress制作百度地图xml,山东省城乡住房和建设厅网站ActiveMQ
安装
下载网址#xff1a;ActiveMQ
一定要和自己安装的jdk版本匹配#xff0c;不然会报错 下载到本地之后解压缩 有可能端口号被占用 解除端口号占用#xff0c;参考#xff1a;Windows_端口被占用
打开cmd
查询所有的端口号
netstat -nao查询指定端口号
n…ActiveMQ
安装
下载网址ActiveMQ
一定要和自己安装的jdk版本匹配不然会报错 下载到本地之后解压缩 有可能端口号被占用 解除端口号占用参考Windows_端口被占用
打开cmd
查询所有的端口号
netstat -nao查询指定端口号
netstat -ano|findstr 5672查询什么程序在占用
tasklist | findstr 4756打开任务管理器看看这个程序 erl.exe是什么进程 Erlang 的执行程序Erlang一种编程语言多用于并发和分布式系统现在最广泛使用在消息队列里面。 安装RabbitMQ的好像就是安装的这个看来是RabbitMQ把这个端口给占用掉了换个端口号 将5672改成55672 网址http://localhost:8161/admin/
用户名admin
密码admin 成功。
使用
参考消息队列之 ActiveMQ Java访问ActiveMQ实例 引入依赖 dependencygroupIdorg.apache.activemq/groupIdartifactIdactivemq-all/artifactIdversion5.15.2/version/dependency消息生产者
package mq.activeMQ;import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;public class TopicPublisher {/*** 默认用户名*/public static final String USERNAME ActiveMQConnection.DEFAULT_USER;/*** 默认密码*/public static final String PASSWORD ActiveMQConnection.DEFAULT_PASSWORD;/*** 默认连接地址*/public static final String BROKER_URL ActiveMQConnection.DEFAULT_BROKER_URL;public static void main(String[] args) {//创建连接工厂ConnectionFactory connectionFactory new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKER_URL);try {//创建连接Connection connection connectionFactory.createConnection();//开启连接connection.start();//创建会话不需要事务Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//创建 Topic用作消费者订阅消息Topic myTestTopic session.createTopic(activemq-topic-test1);//消息生产者MessageProducer producer session.createProducer(myTestTopic);for (int i 1; i 3; i) {TextMessage message session.createTextMessage(发送消息 i);producer.send(myTestTopic, message);}//关闭资源session.close();connection.close();} catch (JMSException e) {e.printStackTrace();}}
}消息消费者
package mq.activeMQ;import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;public class TopicSubscriber {/*** 默认用户名*/public static final String USERNAME ActiveMQConnection.DEFAULT_USER;/*** 默认密码*/public static final String PASSWORD ActiveMQConnection.DEFAULT_PASSWORD;/*** 默认连接地址*/public static final String BROKER_URL ActiveMQConnection.DEFAULT_BROKER_URL;public static void main(String[] args) {//创建连接工厂ConnectionFactory connectionFactory new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKER_URL);try {//创建连接Connection connection connectionFactory.createConnection();//开启连接connection.start();//创建会话不需要事务Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//创建 TopicTopic myTestTopic session.createTopic(activemq-topic-test1);MessageConsumer messageConsumer session.createConsumer(myTestTopic);messageConsumer.setMessageListener(new MessageListener() {Overridepublic void onMessage(Message message) {try {System.out.println(消费者1 接收到消息 ((TextMessage) message).getText());} catch (JMSException e) {e.printStackTrace();}}});MessageConsumer messageConsumer2 session.createConsumer(myTestTopic);messageConsumer2.setMessageListener(new MessageListener() {Overridepublic void onMessage(Message message) {try {System.out.println(消费者2 接收到消息 ((TextMessage) message).getText());} catch (JMSException e) {e.printStackTrace();}}});MessageConsumer messageConsumer3 session.createConsumer(myTestTopic);messageConsumer3.setMessageListener(new MessageListener() {Overridepublic void onMessage(Message message) {try {System.out.println(消费者3 接收到消息 ((TextMessage) message).getText());} catch (JMSException e) {e.printStackTrace();}}});//让主线程休眠100秒使消息消费者对象能继续存活一段时间从而能监听到消息Thread.sleep(100 * 1000);//关闭资源session.close();connection.close();} catch (Exception e) {e.printStackTrace();}}
}
giteeJAVA集成AcitveMQ
启动ActiveMQ服务器 启动 Spring整合ActiveMQ
引入依赖
dependencygroupIdorg.apache.activemq/groupIdartifactIdactivemq-all/artifactIdversion5.15.2/version
/dependencydependencygroupIdorg.springframework/groupIdartifactIdspring-jms/artifactIdversion4.3.10.RELEASE/version
/dependencydependencygroupIdorg.apache.activemq/groupIdartifactIdactivemq-pool/artifactIdversion5.15.0/version
/dependency
Spring配置文件
?xml version1.0 encodingUTF-8?
beans xmlnshttp://www.springframework.org/schema/beansxmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexmlns:contexthttp://www.springframework.org/schema/contextxsi:schemaLocationhttp://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans-3.0.xsdhttp://www.springframework.org/schema/contexthttp://www.springframework.org/schema/context/spring-context-3.0.xsd!--这里要改base-package基础包--!--下面还有要改的看注释--context:component-scan base-packagecom.example.spring_activemq.activeMQ/!--连接池--bean idjmsFactory classorg.apache.activemq.pool.PooledConnectionFactory destroy-methodstopproperty nameconnectionFactorybean classorg.apache.activemq.ActiveMQConnectionFactoryproperty namebrokerURLvaluetcp://localhost:61616/value/property/bean/propertyproperty namemaxConnections value100/property/bean!--缓存--bean idcachingConnectionFactory classorg.springframework.jms.connection.CachingConnectionFactoryproperty nametargetConnectionFactory refjmsFactory/property namesessionCacheSize value1//beanbean idjmsTemplate classorg.springframework.jms.core.JmsTemplate!--获取连接、会话等对象--property nameconnectionFactory refcachingConnectionFactory/!--消息转换器--property namemessageConverterbean classorg.springframework.jms.support.converter.SimpleMessageConverter//property/bean!--实际发送和接收消息的目的地--bean idtestQueue classorg.apache.activemq.command.ActiveMQQueueconstructor-arg namename valuespring-queue//beanbean idtestTopic classorg.apache.activemq.command.ActiveMQTopicconstructor-arg index0 valuespring-topic//bean!--这里也要改class--!--队列消息下的监视器--bean idqueueListener classcom.example.spring_activemq.activeMQ.QueueListener/!--主题模式下的接收器--bean idtopic1Listener classcom.example.spring_activemq.activeMQ.Topic1Listener/bean idtopic2Listener classcom.example.spring_activemq.activeMQ.Topic2Listener/!--将消息监视器绑定到具体的消息目的地上--bean idqueueContainerclassorg.springframework.jms.listener.DefaultMessageListenerContainerproperty nameconnectionFactory refcachingConnectionFactory/property namedestination reftestQueue/property namemessageListener refqueueListener//beanbean idtopic1Containerclassorg.springframework.jms.listener.DefaultMessageListenerContainer!--缓存--property nameconnectionFactory refcachingConnectionFactory/!--目的地--property namedestination reftestTopic/!--监视器--property namemessageListener reftopic1Listener//beanbean idtopic2Containerclassorg.springframework.jms.listener.DefaultMessageListenerContainerproperty nameconnectionFactory refcachingConnectionFactory/property namedestination reftestTopic/property namemessageListener reftopic2Listener//bean/beans
消息服务类
package com.example.spring_activemq.activeMQ;import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;import javax.annotation.Resource;
import javax.jms.*;Service
public class MessageService {Resource(name jmsTemplate)private JmsTemplate jmsTemplate;Resource(name testQueue)private Destination testQueue;Resource(name testTopic)private Destination testTopic;//向队列发送消息public void sendQueueMessage(String messageContent) {jmsTemplate.send(testQueue, new MessageCreator() {Overridepublic Message createMessage(Session session) throws JMSException {TextMessage msg session.createTextMessage();// 设置消息内容msg.setText(messageContent);return msg;}});}//向主题发送消息public void sendTopicMessage(String messageContent) {jmsTemplate.send(testTopic, new MessageCreator() {Overridepublic Message createMessage(Session session) throws JMSException {TextMessage msg session.createTextMessage();// 设置消息内容msg.setText(messageContent);return msg;}});}
}
消息监听器类
队列监听器
package com.example.spring_activemq.activeMQ;import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;public class QueueListener implements MessageListener {Overridepublic void onMessage(Message message) {if (message instanceof TextMessage) {try {TextMessage txtMsg (TextMessage) message;String messageStr txtMsg.getText();System.out.println(队列监听器接收到文本消息 messageStr);} catch (JMSException e) {e.printStackTrace();}} else {throw new IllegalArgumentException(只支持 TextMessage 类型消息);}}
}
订阅消息监听器
package com.example.spring_activemq.activeMQ;import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;public class Topic1Listener implements MessageListener {Overridepublic void onMessage(Message message) {if (message instanceof TextMessage) {try {TextMessage txtMsg (TextMessage) message;String messageStr txtMsg.getText();System.out.println(主题监听器1 接收到文本消息 messageStr);} catch (JMSException e) {e.printStackTrace();}} else {throw new IllegalArgumentException(只支持 TextMessage 类型消息);}}
}
package com.example.spring_activemq.activeMQ;import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;public class Topic2Listener implements MessageListener {Overridepublic void onMessage(Message message) {if (message instanceof TextMessage) {try {TextMessage txtMsg (TextMessage) message;String messageStr txtMsg.getText();System.out.println(主题监听器2 接收到文本消息 messageStr);} catch (JMSException e) {e.printStackTrace();}} else {throw new IllegalArgumentException(只支持 TextMessage 类型消息);}}
}
启动应用
package com.example.spring_activemq.activeMQ;import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;public class StartApplication {public static void main(String[] args) {//这里总是显示空指针找半天也没找出来哪里错了不找了等以后学学再找。ApplicationContext ctx new ClassPathXmlApplicationContext(classpath:spring-context.xml);MessageService messageService (MessageService) ctx.getBean(messageService);messageService.sendQueueMessage(我的测试消息1);messageService.sendTopicMessage(我的测试消息2);messageService.sendTopicMessage(我的测试消息3);}}
giteeSpring集成ActiveMQ
