消息队列工具 ActiveMQ[通俗易懂]

消息队列工具 ActiveMQ[通俗易懂]1 、简介同类产品: RabbitMQ 、 Kafka、Redis(List)1.1 对比RabbitMQ最接近的同类型产品,经常拿来比较,性能

大家好,欢迎来到IT知识分享网。

1 、简介

消息队列工具 ActiveMQ[通俗易懂]

同类产品: RabbitMQ 、 Kafka、Redis(List)

1.1 对比RabbitMQ

最接近的同类型产品,经常拿来比较,性能伯仲之间,基本上可以互相替代。最主要区别是二者的协议不同RabbitMQ的协议是AMQP(Advanced Message Queueing Protoco),而ActiveMQ使用的是JMS(Java Messaging Service )协议。顾名思义JMS是针对Java体系的传输协议,队列两端必须有JVM,所以如果开发环境都是java的话推荐使用ActiveMQ,可以用Java的一些对象进行传递比如Map、BLob、Stream等。而AMQP通用性较强,非java环境经常使用,传输内容就是标准字符串。

另外一点就是RabbitMQ用Erlang开发,安装前要装Erlang环境,比较麻烦。ActiveMQ解压即可用不用任何安装。

1.2 对比KafKa

Kafka性能超过ActiveMQ等传统MQ工具,集群扩展性好。

弊端是:

在传输过程中可能会出现消息重复的情况,

不保证发送顺序

一些传统MQ的功能没有,比如消息的事务功能。

所以通常用Kafka处理大数据日志。

1.3 对比Redis

其实Redis本身利用List可以实现消息队列的功能,但是功能很少,而且队列体积较大时性能会急剧下降。对于数据量不大、业务简单的场景可以使用。

2 安装 ActiveMQ

拷贝apache-activemq-5.14.4-bin.tar.gz到Linux服务器的/opt

解压缩 tar -zxvf apache-activemq-5.14.4-bin.tar.gz

重命名 mv apache-activemq-5.14.4 activemq

vim /opt/activemq/bin/activemq

消息队列工具 ActiveMQ[通俗易懂]

查看java环境:vim /etc/profile 或者 echo $JAVA_HOME

增加两行

JAVA_HOME=”/opt/jdk1.8.0_152″
JAVA_CMD=”/opt/jdk1.8.0_152/bin”

IT知识分享网

注册服务

ln -s /opt/activemq/bin/activemq /etc/init.d/activemq(软连接必须使用绝对路径)

chkconfig –add activemq

# 禁止使用

# cp /opt/activemq/bin/activemq /etc/init.d/activemq

启动服务

service activemq start

消息队列工具 ActiveMQ[通俗易懂]

关闭服务

service activemq stop

通过netstat 查看端口

# netstat -tlnp

t:表示tcp

l:表示监听

n: 将ip和端口转换成域名和服务名

p:显示的程序名

消息队列工具 ActiveMQ[通俗易懂]

activemq两个重要的端口,一个是提供消息队列的默认端口:61616

另一个是控制台端口8161

通过控制台测试

启动消费端

消息队列工具 ActiveMQ[通俗易懂]

进入网页控制台

消息队列工具 ActiveMQ[通俗易懂]

账号/密码默认: admin/admin

点击Queues

消息队列工具 ActiveMQ[通俗易懂]

消息队列工具 ActiveMQ[通俗易懂]

消息队列工具 ActiveMQ[通俗易懂]

观察客户端

消息队列工具 ActiveMQ[通俗易懂]

3 在Java中使用消息队列

3.1 在gmall-service-util中导入依赖坐标

IT知识分享网<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-activemq</artifactId>
   <exclusions>
      <exclusion>
         <groupId>org.slf4j</groupId>
         <artifactId>slf4j-log4j12</artifactId>
      </exclusion>
   </exclusions>
</dependency>
 
<dependency>
   <groupId>org.apache.activemq</groupId>
   <artifactId>activemq-pool</artifactId>
   <version>5.15.2</version>
   <exclusions>
      <exclusion>
         <groupId>org.slf4j</groupId>
         <artifactId>slf4j-log4j12</artifactId>
      </exclusion>
   </exclusions>
</dependency>
 

3.2 在payment项目中添加producer端

 
public class ProducerTest {
    public static void main(String[] args) throws JMSException {
        // 创建连接工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(“tcp://192.168.67.201:61616”);
        Connection connection = connectionFactory.createConnection();
        connection.start();
        // 创建session 第一个参数表示是否支持事务,false时,第二个参数Session.AUTO_ACKNOWLEDGE自动签收,Session.CLIENT_ACKNOWLEDGE手动签收,DUPS_OK_ACKNOWLEDGE 订阅时签收其中一个
        // 第一个参数设置为true时,第二个参数可以忽略 服务器设置为SESSION_TRANSACTED
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 创建队列
        Queue queue = session.createQueue(“Atguigu”);
        MessageProducer producer = session.createProducer(queue);
        // 创建消息对象
        ActiveMQTextMessage activeMQTextMessage = new ActiveMQTextMessage();
        activeMQTextMessage.setText(“hello ActiveMq!”);
        // 发送消息
        producer.send(activeMQTextMessage);
        producer.close();
        connection.close();
    }
}

注意:如果有事务需要先提交事务session.commit();

消息队列工具 ActiveMQ[通俗易懂]

在zookeeper中,有持久化-非持久化。

Number Of Consumers 消费者 这个是消费者端的消费者数量

Messages Enqueued 进入队列的消息 进入队列的总数量,包括出队列的。 这个数量只增不减

Messages Dequeued 出了队列的消息 可以理解为是消费者消费掉的数量

总结:

当有一个消息进入这个队列时,等待消费的消息是1,进入队列的消息是1。
当消息消费后,等待消费的消息是0,进入队列的消息是1,出队列的消息是1.
在来一条消息时,等待消费的消息是1,进入队列的消息就是2.

3.3 在payment项目中添加consumer端

IT知识分享网public class ConsumerTest {
    public static void main(String[] args) throws JMSException {
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, “tcp://192.168.67.201:61616”);
        // 创建连接
        Connection connection = activeMQConnectionFactory.createConnection();
        connection.start();
        // 创建会话
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 创建队列
        Queue queue = session.createQueue(“Atguigu”);
        // 创建Consumer
        MessageConsumer consumer = session.createConsumer(queue);
        // 接收消息
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                // 参数就是收到的消息
                if (message instanceof  TextMessage){
                    try {
                        String text = ((TextMessage) message).getText();
                        System.out.println(text+“接收的消息!”);
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        consumer.close();
        session.close();
    }
}
 

3.4 关于事务控制

消息队列工具 ActiveMQ[通俗易懂]

3.5 持久化与非持久化

通过producer.setDeliveryMode(DeliveryMode.PERSISTENT) 进行设置

持久化的好处就是当activemq宕机的话,消息队列中的消息不会丢失。非持久化会丢失。但是会消耗一定的性能。

持久化:当服务器宕机,消息依然存在。

非持久化:当服务器宕机,消息不存在。

在zookeeper中,有持久化-非持久化。

了解更多java相关技术内容欢迎关注尚硅谷教育!

免责声明:本站所有文章内容,图片,视频等均是来源于用户投稿和互联网及文摘转载整编而成,不代表本站观点,不承担相关法律责任。其著作权各归其原作者或其出版社所有。如发现本站有涉嫌抄袭侵权/违法违规的内容,侵犯到您的权益,请在线联系站长,一经查实,本站将立刻删除。 本文来自网络,若有侵权,请联系删除,如若转载,请注明出处:https://yundeesoft.com/7048.html

(0)
上一篇 2023-01-03 09:53
下一篇 2023-03-11 17:59

相关推荐

发表回复

您的电子邮箱地址不会被公开。 必填项已用 * 标注

关注微信