ActiveMQ学习笔记_随笔

ActiveMQ学习笔记

1. Active简介2. Active安装以及问题总结

安装问题总结 3. Active基本使用

传输文本信息传输对象消息实现队列服务监听处理消息使用Topic模型SpringBoot整合ActiveMQ

1. Active简介

它是一种传输消息的中间件

2. Active安装以及问题总结安装

访问官方网站选择相应的版本进行下载，地址为http://activemq.apache.org也可以从我的百度网盘获取5.9版本的https://pan.baidu.com/s/19k2W34u7jcBffiHfG3u8IQ
提取码：45lt

这里选择的是在CentOs7.6上安装ActiveMQ的5.9的版本

使用xftp将.gz文件上传到linux上解压压缩包到当前文件夹
tar -zxf apache-activemq-5.9.0-bin.tar.gz进入bin目录中
cd apache-activemq-5.9.0/bin/启动它(后面也可以跟 restart、status、stop)
./activemq start 问题总结

在bin目录下执行 ./activemq conaole 查看日志信息

在启动active时，启动失败，报错如下：

(you can configure options in one of these file: /etc/default/activemq /root/.activemqrc)

INFO: Invoke the following command to create a configuration file
/apache-activemq-5.10.2/bin/activemq setup [ /etc/default/activemq | /root/.activemqrc ]

INFO: Using java '/jdk/jdk1.8.0_161/bin/java'
INFO: Starting - inspect logfiles specified in logging.properties and log4j.properties to get details

INFO: pidfile created : '/apache-activemq-5.10.2/data/activemq-iZ2zefq9vbrnj5odlyp6g5Z.pid' (pid '16600')

解决办法：在bin目录中执行下面命令：
./activemq setup /etc/default/activemq

启动Active失败，报错如下：

ERROR | Failed to start Apache ActiveMQ 
([localhost, ID:CentOs7.6-35915-1642471492154-0:1], 
java.net.URISyntaxException: Illegal character in hostname at index 13: ws://CentOs7.6:61614?maximumConnections=1000&wireFormat.maxframeSize=104857600)

主机名不合法

解决方法 : 进入与bin同级目录的conf目录中，修改配置文件
vim activemq.xml

3. Active基本使用

pom文件中导入相关坐标


     org.apache.activemq
     activemq-all
     5.9.0

传输文本信息

生产者如下:

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Producer {

    public void sendMessageToActiveMQ(){
        //定义链接工厂
        ConnectionFactory connectionFactory = null;
        //定义链接对象
        Connection connection = null;
        //定义会话对象
        Session session = null;
        //定义目的地
        Destination destination = null;
        //定义消息发送者
        MessageProducer messageProducer = null;
        //定义消息对象
        Message message = null;

        try{
            //参数分别为:访问ActiveMQ服务的用户名、密码(可以通过jetty-ream.properties文件进行修改)、服务的路径地址(协议名://ip地址:端口)
            connectionFactory = new ActiveMQConnectionFactory("admin","admin","tcp://192.168.137.130:61616");
            //创建连接
            connection = connectionFactory.createConnection();
            //开启连接
            connection.start();
            //创建会话
            
            session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
            //创建目的地即队列
            destination = session.createQueue("queue1");
            //创建消息生产者
            messageProducer = session.createProducer(destination);
            //创建消息对象
            message = session.createTextMessage("Hello ActiveMQ");
            //发送消息
            messageProducer.send(message);
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(messageProducer != null){
                try {
                    messageProducer.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
            if(session != null){
                try {
                    session.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
            if(connection != null){
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }

    }

}

消费者如下:

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Consumer {

    public void receiveMessageFromActiveMQ(){
        //定义链接工厂
        ConnectionFactory connectionFactory = null;
        //定义链接对象
        Connection connection = null;
        //定义会话对象
        Session session = null;
        //定义目的地
        Destination destination = null;
        //定义消息接收者
        MessageConsumer messageConsumer = null;
        //定义消息对象
        Message message = null;

        try{
            //参数分别为:访问ActiveMQ服务的用户名、密码(可以通过jetty-ream.properties文件进行修改)、服务的路径地址(协议名://ip地址:端口)
            connectionFactory = new ActiveMQConnectionFactory("admin","admin","tcp://192.168.137.130:61616");
            //创建连接
            connection = connectionFactory.createConnection();
            //开启连接
            connection.start();
            //创建会话
            
            session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
            //指明监控的队列
            destination = session.createQueue("queue1");
            //创建消息消费者
            messageConsumer = session.createConsumer(destination);
            //拿到消息对象
            message = messageConsumer.receive();

            System.out.println("消息为:" + ((TextMessage)message).getText());

        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(messageConsumer != null){
                try {
                    messageConsumer.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
            if(session != null){
                try {
                    session.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
            if(connection != null){
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }

    }

}

传输对象消息

生产者如下:

只需修改上面的相关代码对象必须实现序列化接口这里传输的是User对象

//创建消息对象
//message = session.createTextMessage("Hello ActiveMQ");
  message = session.createObjectMessage(new User(15,"111"));

消费者如下:

修改上面消费者的相关代码

message = messageConsumer.receive();
//System.out.println("消息为:" + ((TextMessage)message).getText());
ObjectMessage objectMessage = (ObjectMessage) message;
User user = (User) objectMessage.getObject();
System.out.println(user);

实现队列服务监听处理消息

在上面消费者的代码中去掉获取消息以及打印消息的部分，添加如下代码

messageConsumer.setMessageListener(new MessageListener() {
             public void onMessage(Message message) {
                try {
                    //里面就是在处理消息，这里处理的是对象消息
                     ObjectMessage objectMessage = (ObjectMessage) message;
                     User user = (User) objectMessage.getObject();
                     System.out.println(user);
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });

注意去掉finally，因为监听的时候，会话对象，连接对象必须存在，不能关闭掉

使用Topic模型

//将生产者以及消费者中的代码修改为最后一句
//destination = session.createQueue("queue1");
destination = session.createTopic("topic1");

这样就可以让多个消费者监视一个topic了

SpringBoot整合ActiveMQ

1.导入坐标依赖


    org.springframework.boot
    spring-boot-starter-activemq



    org.apache.activemq
    activemq-pool
    5.15.0

2.配置如下

#ActiveMQ服务地址
spring.activemq.broker-url=tcp://192.168.137.130:61616
#true表示使用内置的MQ，false则连接服务器
spring.activemq.in-memory=false
#true表示使用连接池；false时，每发送一条数据创建一个连接
spring.activemq.pool.enabled=true
#连接池最大连接数
spring.activemq.pool.max-connections=10
#空闲的连接过期时间，默认为30秒
spring.activemq.pool.idle-timeout=30000
#强制的连接过期时间
spring.activemq.pool.expiry-timeout=0

3.在启动类上添加@EnableJms注解开启消息队列

4.配置队列：

import javax.jms.Queue;
import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ActiveMQConfig {
    //定义存放消息的队列
    @Bean
    public Queue queue() {
        return new ActiveMQQueue("Queue111");
    }
}

5.生产者如下:

import javax.jms.Queue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.Controller;
 
@Controller
public class ProviderController {
 
    //注入之前配置的队列队列
    @Autowired
    private Queue queue;
 
    //注入springboot封装的工具类
    @Autowired
    private JmsMessagingTemplate template;
 
    @RequestMapping("test")
    public void sendMessage() {
        //方法一：添加消息到消息队列
        template.convertAndSend(queue, "Hello ActiveMQ");
        //方法二：这种方式不需要手动创建queue，系统会自行创建名为test的队列
        //template.convertAndSend("test", "Hello ActiveMQ");
    }
}

6.消费者如下:

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Component;
 
@Component
public class ConsumerService {
 
    @Autowired
    private JmsMessagingTemplate jmsMessagingTemplate;
 
    // 使用JmsListener配置消费者监听的队列
    @JmsListener(destination = "Queue111")
    public void handleMessage(String msg) {
        System.out.println("消息为:" + msg);
    }
}

欢迎分享，转载请注明来源：内存溢出

原文地址: http://outofmemory.cn/zaji/5709186.html

ActiveMQ学习笔记

发表评论

评论列表（0条）