RabbitMQ工作模式1 Hello world工作模式

简介: RabbitMQ工作模式1 Hello world工作模式

RabbitMQ工作模式1 Hello world工作模式

使用Java代码操作RabbitMQ

需求:使用简单模式完成消息传递,简单模式如下所示,P代表生产者,C代表消费者,中间是队列

六种模式中最简单的就是它(工作模式)

模式说明

image.png


实现步骤(接下来的工作模式也会安装相同的代码进行修改)

1 创建两个工程,生产者(P)和消费者(C)

创建一个新的项目

image.png

在项目中创建两个新的模块 分别为生产者和消费者


2 分别引入对应工程的依赖

他俩的pom分别引入

<dependencies>
<!--rabbitMQ依赖 Java客户端-->
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.6.0</version>
    </dependency>
</dependencies>
<!--编译插件  -->
<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.8.0</version>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
            </configuration>
        </plugin>
    </plugins>
</build>
3 编写生产者发送消息
package com.wyh.producer;
/**
 * @program: SpringBoot-RabbitMQ
 * @description: RabbitMQ生产者
 * @author: 魏一鹤
 * @createDate: 2022-03-23 22:40
 **/
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
 *  producer主要用来发送消息
 *
 *
**/
public class HelloWorld {
public static void main(String[] args) throws IOException, TimeoutException {
//1.建立工厂  一般连接都是通过连接工厂进行连接 所以要创建连接工厂
        ConnectionFactory factory=new ConnectionFactory();
//2 设置参数 比如说虚拟机 用户名 ip 密码 端口等 当然不设置也有默认参数
        factory.setHost("127.0.0.1  ");//设置主机ip  172.16.98.133是远程的服务 就是rabbitMQ服务页面的ip 如果不设置默认值为localhost (127.0.0.1)
        factory.setPort(5672);//设置端口 默认值也是5672
        factory.setVirtualHost("/itcast_wyh");//设置虚拟机 默认值/ 杠
        factory.setUsername("weiyihe");//设置用户名 默认值 guest 游客
        factory.setPassword("weiyihe");//设置密码 默认值 guest 游客
        //3 创建连接Connection
        Connection connection = factory.newConnection();
//4 创建channel
        Channel channel = connection.createChannel();
//5 创建队列Queue 它的参数比较多 下面一一说明
        /**
         *  1 String queue, 队列名称
         *  2 boolean durable,   是否持久化,当MQ重启之后还在 持久化到数据库中
         *  3 boolean exclusive, 它有两个功能 1是否独占:只能有一个消费者监听这个队列 2当connection关闭时,是否删除队列.一般设置为false
         *  4 boolean autoDelete, 是否自动删除,让没有consumer时,会自动删除掉
         *  5 Map<String, Object> arguments  参数信息
         *
        **/
        //如果没有一个名字叫hello_world的队列,则会创建该队列,如果有则不会创建
        AMQP.Queue.DeclareOk queue = channel.queueDeclare("hello_world", true, false, false, null);
//6 发送消息 它的参数比较多 下面一一说明
        //发送的消息队列  用来发送给消费者 一般发送的消息都是字节
        String body="hello RabbitMQ!";
//   1 String exchange,交换机的名称,简单模式下交换机会使用默认的 ""
        //   2 String routingKey, 路由名称
        //   3 BasicProperties props, 配置信息
        //   4 byte[] body  真实的发送的消息数据
        channel.basicPublish("","hello_world",null,body.getBytes());
//7 释放资源
        channel.close();
        connection.close();
    }
}
4 编写消费者接收消息
package com.wyh.consumer;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
 * @program: SpringBoot-RabbitMQ
 * @description: RabbitMQ消费者Consumer
 * @author: 魏一鹤
 * @createDate: 2022-03-24 22:08
 **/
/**
 *  consumer主要用来消费消息
 *
 *
 **/
public class HelloWorld_consumer {
public static void main(String[] args) throws IOException, TimeoutException {
//1.建立工厂  一般连接都是通过连接工厂进行连接 所以要创建连接工厂
        ConnectionFactory factory=new ConnectionFactory();
//2 设置参数 比如说虚拟机 用户名 ip 密码 端口等 当然不设置也有默认参数
        factory.setHost("127.0.0.1");//设置主机ip  172.16.98.133是远程的服务 就是rabbitMQ服务页面的ip 如果不设置默认值为localhost (127.0.0.1)
        factory.setPort(5672);//设置端口 默认值也是5672
        factory.setVirtualHost("/itcast_wyh");//设置虚拟机 默认值/ 杠
        factory.setUsername("weiyihe");//设置用户名 默认值 guest 游客
        factory.setPassword("weiyihe");//设置密码 默认值 guest 游客
        //3 创建连接Connection
        Connection connection = factory.newConnection();
//4 创建channel
        Channel channel = connection.createChannel();
//5 创建队列Queue 它的参数比较多 下面一一说明
        /**
         *  1 String queue, 队列名称
         *  2 boolean durable,   是否持久化,当MQ重启之后还在 持久化到数据库中
         *  3 boolean exclusive, 它有两个功能 1是否独占:只能有一个消费者监听这个队列 2当connection关闭时,是否删除队列.一般设置为false
         *  4 boolean autoDelete, 是否自动删除,让没有consumer时,会自动删除掉
         *  5 Map<String, Object> arguments  参数信息
         *
         **/
        //如果没有一个名字叫hello_world的队列,则会创建该队列,如果有则不会创建
        AMQP.Queue.DeclareOk queue = channel.queueDeclare("hello_world", true, false, false, null);
//6 接收消息  它的参数比较多 下面一一说明
        //  String queue, 队列名称
        //  boolean autoAck, 是否自动确认  消费者收到消息会自动告诉MQ它收到了消息
        //  Consumer callback  回调对象 可以监听一些方法
        //consumer本质是一个接口 需要创建它的实现类
        Consumer consumer=new DefaultConsumer(channel){
//匿名内部类 重写它的方法
            //回调方法 当收到消息后,会自动执行该方法 它有一些参数
            //String consumerTag 标识
            //Envelope envelope 可以获取一些信息 比如交换机 路由key
            //AMQP.BasicProperties properties 配置信息
            // byte[] body 数据
            //
            @Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("consumerTag = " + consumerTag);
                System.out.println("exchange = " + envelope.getExchange());
                System.out.println("routingKey = " + envelope.getRoutingKey());
                System.out.println("properties = " + properties);
                System.out.println("body = " + new String(body));
            }
/**   打印的信息
             consumerTag = amq.ctag-7ojIfmsjb9wcpbkwzwEobw
             exchange = 
             routingKey = hello_world
             properties = #contentHeader<basic>(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
             body = hello RabbitMQ!
             consumerTag = amq.ctag-7ojIfmsjb9wcpbkwzwEobw
             exchange = 
             routingKey = hello_world
             properties = #contentHeader<basic>(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
             body = hello RabbitMQ 222!
            **/
        };
        channel.basicConsume("hello_world",true,consumer);
//消费者本质是一个监听 所以不要去关闭资源
    }
}

consumerTag = amq.ctag-7ojIfmsjb9wcpbkwzwEobw

exchange =

routingKey = hello_world

properties = #contentHeader<basic>(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)

body = hello RabbitMQ!

consumerTag = amq.ctag-7ojIfmsjb9wcpbkwzwEobw

exchange =

routingKey = hello_world

properties = #contentHeader<basic>(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)

body = hello RabbitMQ 222!

5 测试

现在是没有队列的


image.png


运行程序代码,发现有了一个队列


image.png


image.png


package com.wyh.producer;
/**
 * @program: SpringBoot-RabbitMQ
 * @description: RabbitMQ生产者
 * @author: 魏一鹤
 * @createDate: 2022-03-23 22:40
 **/
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
 *  producer主要用来发送消息
 *
 *
**/
public class HelloWorld {
public static void main(String[] args) throws IOException, TimeoutException {
//1.建立工厂  一般连接都是通过连接工厂进行连接 所以要创建连接工厂
        ConnectionFactory factory=new ConnectionFactory();
//2 设置参数 比如说虚拟机 用户名 ip 密码 端口等 当然不设置也有默认参数
        factory.setHost("127.0.0.1");//设置主机ip  172.16.98.133是远程的服务 就是rabbitMQ服务页面的ip 如果不设置默认值为localhost (127.0.0.1)
        factory.setPort(5672);//设置端口 默认值也是5672
        factory.setVirtualHost("/itcast_wyh");//设置虚拟机 默认值/ 杠
        factory.setUsername("weiyihe");//设置用户名 默认值 guest 游客
        factory.setPassword("weiyihe");//设置密码 默认值 guest 游客
        //3 创建连接Connection
        Connection connection = factory.newConnection();
//4 创建channel
        Channel channel = connection.createChannel();
//5 创建队列Queue 它的参数比较多 下面一一说明
        /**
         *  1 String queue, 队列名称
         *  2 boolean durable,   是否持久化,当MQ重启之后还在 持久化到数据库中
         *  3 boolean exclusive, 它有两个功能 1是否独占:只能有一个消费者监听这个队列 2当connection关闭时,是否删除队列.一般设置为false
         *  4 boolean autoDelete, 是否自动删除,让没有consumer时,会自动删除掉
         *  5 Map<String, Object> arguments  参数信息
         *
        **/
        //如果没有一个名字叫hello_world的队列,则会创建该队列,如果有则不会创建
        AMQP.Queue.DeclareOk queue = channel.queueDeclare("hello_world", true, false, false, null);
//6 发送消息 它的参数比较多 下面一一说明
        //发送的消息队列  用来发送给消费者 一般发送的消息都是字节
        String body="hello RabbitMQ!";
//   1 String exchange,交换机的名称,简单模式下交换机会使用默认的 ""
        //   2 String routingKey, 路由名称
        //   3 BasicProperties props, 配置信息
        //   4 byte[] body  真实的发送的消息数据
        channel.basicPublish("","hello_world",null,body.getBytes());
//7 释放资源
        //channel.close();
        //connection.close();
    }
}

可以发现,连接和channel都有了,而且队列多了一条信息,因为我们又执行了一次

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
2月前
|
消息中间件
RabbitMQ的 RPC 消息模式你会了吗?
【9月更文挑战第11天】RabbitMQ 的 RPC(远程过程调用)消息模式允许客户端向服务器发送请求并接收响应。其基本原理包括:1) 客户端发送请求,创建回调队列并设置关联标识符;2) 服务器接收请求并发送响应至回调队列;3) 客户端根据关联标识符接收并匹配响应。实现步骤涵盖客户端和服务器的连接、信道创建及请求处理。注意事项包括关联标识符唯一性、回调队列管理、错误处理及性能考虑。RPC 模式适用于构建可靠的分布式应用程序,但需根据需求调整优化。
|
3月前
|
消息中间件 开发者
【RabbitMQ深度解析】Topic交换器与模式匹配:掌握消息路由的艺术!
【8月更文挑战第24天】在消息队列(MQ)体系中,交换器作为核心组件之一负责消息路由。特别是`topic`类型的交换器,它通过模式匹配实现消息的精准分发,适用于发布-订阅模式。不同于直接交换器和扇形交换器,`topic`交换器支持更复杂的路由策略,通过带有通配符(如 * 和 #)的模式字符串来定义队列与交换器间的绑定关系。
70 2
|
3月前
|
消息中间件
RabbitMQ广播模式
RabbitMQ广播模式
60 1
|
3月前
|
消息中间件 应用服务中间件 网络安全
rabbitMQ镜像模式搭建
rabbitMQ镜像模式搭建
|
4月前
|
消息中间件 传感器 负载均衡
消息队列 MQ使用问题之如何配置一主一从的同步复制模式
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
消息队列 MQ使用问题之如何配置一主一从的同步复制模式
|
4月前
|
消息中间件 存储 Kafka
MetaQ/RocketMQ 原理问题之RocketMQ DLedger融合模式的问题如何解决
MetaQ/RocketMQ 原理问题之RocketMQ DLedger融合模式的问题如何解决
|
3月前
|
消息中间件 Java Maven
RabbitMQ通配符模式
RabbitMQ通配符模式
60 0
|
4月前
|
消息中间件 Java Apache
消息队列 MQ使用问题之如何在内外网环境下使用单组节点单副本模式
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
4月前
|
消息中间件 负载均衡 RocketMQ
MetaQ/RocketMQ 原理问题之在广播模式下,RebalanceService工作的问题如何解决
MetaQ/RocketMQ 原理问题之在广播模式下,RebalanceService工作的问题如何解决
|
5月前
|
消息中间件
RabbitMQ配置单活模式队列
RabbitMQ配置单活模式队列
137 0