使用RabbitMQ

使用RabbitMQ

1 Docker安装RabbitMQ

1.1 安装RabbitMQ

# 下载含有管理页面的镜像
docker pull rabbitmq:3.8.8-management

# 创建容器
# 5672：应用访问端口；15672：控制台Web端口号；
docker run -itd \
--name=my-rabbitmq \
--restart=always \
-p 15672:15672 \
-p 5672:5672 \
-e RABBITMQ_DEFAULT_USER=rabbitmq \
-e RABBITMQ_DEFAULT_PASS=123456 \
-v /home/rabbitmq/data:/var/lib/rabbitmq rabbitmq:3.8.8-management

1.2 访问RabbitMQ

# 访问网页
http://192.168.108.200:15672

2 SpringBoot使用RabbitMQ

2.1 说明

1. 引入“spring-boot-starter-amqp”依赖；
2. 使用“@Configuration”配置消息队列；
3. 使用RabbitTemplate发送消息；
4. 使用@RabbitListener监听消息；

2.2 工程目录

2.3 pom.xml

<?xml version="1.0" encoding="UTF-8"?><projectxmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.mason</groupId><artifactId>myrabbitmq</artifactId><version>1.0</version><properties><java.version>11</java.version><spring-boot-version>2.3.12.RELEASE</spring-boot-version><maven.compiler.source>11</maven.compiler.source><maven.compiler.target>11</maven.compiler.target></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- Config the RebbitMQ --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency></dependencies><dependencyManagement><!-- 注意：尽量在此配置Spring-Boot版本，子应用中可以不配置SpringBoot版本 --><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-dependencies</artifactId><version>${spring-boot-version}</version><type>pom</type><scope>import</scope></dependency></dependencies></dependencyManagement><build><plugins><!-- 配置Maven插件 --><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.1</version><configuration><source>${java.version}</source><target>${java.version}</target><encoding>UTF-8</encoding></configuration></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-surefire-plugin</artifactId><version>2.22.0</version><configuration><skipTests>true</skipTests></configuration></plugin><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><version>${spring-boot-version}</version></plugin></plugins></build></project>

2.4 yml

server:port:8081servlet:context-path: /myrm

spring:# 设置应用名application:name: my-rabbitmq

  # 配置RabbitMQrabbitmq:host: 192.168.108.200
    port:5672username: rabbitmq
    password:123456

2.5 config

ConfigRabbitmq.java

packagecom.mason.myrabbitmq.config;importorg.springframework.amqp.core.Queue;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importstaticcom.mason.myrabbitmq.config.MyRoutingKey.QUEUE_HELLO;@ConfigurationpublicclassConfigRabbitmq{@BeanpublicQueuecreateDiscoverQueue(){returnnewQueue(QUEUE_HELLO,true);}}

MyRoutingKey.java

packagecom.mason.myrabbitmq.config;publicclassMyRoutingKey{publicstaticfinalStringQUEUE_HELLO="queue_hello";}

2.6 controller

QueueController.java

package com.mason.myrabbitmq.controller;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import static com.mason.myrabbitmq.config.MyRoutingKey.QUEUE_HELLO;

/**
 * " @RestController是@Controller和@ResponseBody和合并 "
 */

@RestController
@RequestMapping("/test")
public class QueueController {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @RequestMapping("/send")
    public String sendMessage() {
        System.out.println("1 发送数据");
        System.out.println("Hello My RabbitMQ");
        this.rabbitTemplate.convertAndSend(QUEUE_HELLO, "Hello My RabbitMQ ");
        return "Success";
    }
}

2.7 service

QueueService.java

packagecom.mason.myrabbitmq.service;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Service;importstaticcom.mason.myrabbitmq.config.MyRoutingKey.QUEUE_HELLO;@ServicepublicclassQueueService{// 监听RabbitMQ@RabbitListener(queues =QUEUE_HELLO)publicvoidreceiveMessage(String msg){System.out.println("2 获取RabbitMQ数据");System.out.println(msg);}}

2.8 MyrabbitmqApplication.java

package com.mason.myrabbitmq;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class MyrabbitmqApplication {
    public static void main(String[] args) {
        SpringApplication.run(MyrabbitmqApplication.class, args);
    }
}

2.9 截图

3 Python使用RabbitMQ

3.1 安装pika

# pika是操作RabbitMQ的依赖包
pip install pika

3,2 工程目录

3.3 main.py

# 导入channelfrom my_decorator import channel
from my_rabbitmq import init_queue

init_queue()# 运行APPif __name__ =='__main__':print("RabbitMQ start consuming")
    channel.start_consuming()

3.4 my_decorator.py

# 导入pika使用RabbitMQimport pika
from pika.adapters.blocking_connection import BlockingChannel

# 导入routing keyfrom my_routing_key import RoutingKey

# 连接RabbitMQdefconn_rabbitmq()-> BlockingChannel:

    host ="192.168.108.200"
    username ="rabbitmq"
    password ="123456"
    port =5672# 配置rabbitmq连接
    crenditials = pika.PlainCredentials(username=username, password=password)
    conn_parameter = pika.ConnectionParameters(host=host, port=port, credentials=crenditials)# 连接rabbitmq
    connection = pika.BlockingConnection(conn_parameter)# Generate the channelreturn connection.channel()# 连接channel
channel = conn_rabbitmq()# 声明 routing keydefdeclare_routing_key():
    channel.queue_declare(RoutingKey.my_routing_key)

declare_routing_key()# 自定义装饰器defqueue(queue_name):"""
    监听rabbitmq消息
    :param queue_name: rabbitmq的消息名称
    :return: 装饰器方法
    """defdecorator(func):
        channel.basic_consume(queue=queue_name, auto_ack=True, on_message_callback=func)return decorator

3.5 my_rabbitmq.py

# 使用装饰器监听rabbitmqfrom my_decorator import queue

# 导入routing keyfrom my_routing_key import RoutingKey

definit_queue():print("初始化QueueMessage")pass@queue(RoutingKey.my_routing_key)defget_message(ch, method, properties, body):print("2 接收RabbitMQ")print(body)pass

3.6 my_routing_key.py

# 设置RoutingKey

class RoutingKey():
    my_routing_key = "my_routing_key"

3.7 my_send_message.py

from my_decorator import channel

# 导入routing key
from my_routing_key import RoutingKey

def send_data():
    print("1 发送RabbitMQ")
    body = "Hello my data"
    channel.basic_publish(exchange='', routing_key=RoutingKey.my_routing_key, body=body)

send_data()

3.8 截图

1. 运行main.py，实时接收信息;
2. 运行my_send_message.py发送消息；