首页 > 资讯 > 服务器 >docker启动rabbitmq以及使用方式详解

682

分享到

docker启动rabbitmq以及使用方式详解

docker运行rabbitmq rabbitmq docker 启动rabbitmq命令 2022-11-13 14:11:33 682人浏览安东尼

摘要

目录搜索RabbitMQ镜像下载镜像启动容器打印容器访问RabbitMQ Management编写生产者类消费者工作队列RabbitMqUtils工具类启动2个工作线程启动发送线程消

搜索rabbitmq镜像

Docker search rabbitmq:management

下载镜像

docker pull rabbitmq:management

启动容器

docker run -d --hostname localhost --name rabbitmq -p 15672:15672 -p 5672:5672 rabbitmq:management

打印容器

docker logs rabbitmq

访问RabbitMQ Management

Http://localhost:15672

账户密码默认:guest

编写生产者类

package com.xun.rabbitmqdemo.example;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Producer {
    private final static String QUEUE_NAME = "hello";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setVirtualHost("/");

        Connection connection = factory.newConnection();

        Channel channel = connection.createChannel();
        
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        String message = "Hello world!";
        
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
        System.out.println(" [x] Sent '"+message+"'");

        channel.close();
        connection.close();
    }
}

运行该方法，可以看到控制台的打印

name=hello的队列收到Message

消费者

package com.xun.rabbitmqdemo.example;

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Receiver {
    private final static String QUEUE_NAME = "hello";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setConnectionTimeout(600000);//milliseconds
        factory.setRequestedHeartbeat(60);//seconds
        factory.setHandshakeTimeout(6000);//milliseconds
        factory.setRequestedChannelMax(5);
        factory.setNetworkRecoveryInterval(500);

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        System.out.println("Waiting for messages. ");

        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println(" [x] Received '" + message + "'");
            }
        };
        channel.basicConsume(QUEUE_NAME,true,consumer);
    }
}

工作队列

RabbitMqUtils工具类

package com.xun.rabbitmqdemo.utils;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class RabbitMqUtils {
    public static Channel getChannel() throws Exception{
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        return channel;
    }
}

启动2个工作线程

package com.xun.rabbitmqdemo.workQueue;

import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.xun.rabbitmqdemo.utils.RabbitMqUtils;

public class Work01 {
    private static final String QUEUE_NAME = "hello";
    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        DeliverCallback deliverCallback = (consumerTag,delivery)->{
            String receivedMessage = new String(delivery.getBody());
            System.out.println("接收消息:"+receivedMessage);
        };
        CancelCallback cancelCallback = (consumerTag)->{
            System.out.println(consumerTag+"消费者取消消费接口回调逻辑");
        };
        System.out.println("C1 消费者启动等待消费....");
        
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
}

package com.xun.rabbitmqdemo.workQueue;

import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.xun.rabbitmqdemo.utils.RabbitMqUtils;

public class Work02 {
    private static final String QUEUE_NAME = "hello";
    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        DeliverCallback deliverCallback = (consumerTag,delivery)->{
            String receivedMessage = new String(delivery.getBody());
            System.out.println("接收消息:"+receivedMessage);
        };
        CancelCallback cancelCallback = (consumerTag)->{
            System.out.println(consumerTag+"消费者取消消费接口回调逻辑");
        };
        System.out.println("C2 消费者启动等待消费....");
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
}

启动工作线程

启动发送线程

package com.xun.rabbitmqdemo.workQueue;

import com.rabbitmq.client.Channel;
import com.xun.rabbitmqdemo.utils.RabbitMqUtils;
import java.util.Scanner;

public class Task01 {
    private static final String QUEUE_NAME = "hello";
    public static void main(String[] args) throws Exception{
        try(Channel channel= RabbitMqUtils.getChannel();){
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
            //从控制台接收消息
            Scanner scanner = new Scanner(System.in);
            while(scanner.hasNext()){
                String message = scanner.next();
                channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
                System.out.println("发送消息完成:"+message);
            }
        }
    }
}

启动发送线程，此时发送线程等待键盘输入

发送4个消息

可以看到2个工作线程按照顺序分别接收message。

消息应答机制

rabbitmq将message发送给消费者后，就会将该消息标记为删除。

但消费者在处理message过程中宕机，会导致消息的丢失。

因此需要设置手动应答。

生产者

import com.xun.rabbitmqdemo.utils.RabbitMqUtils;
import java.util.Scanner;

public class Task02 {
    private static final String TASK_QUEUE_NAME = "ack_queue";
    public static void main(String[] args) throws Exception{
        try(Channel channel = RabbitMqUtils.getChannel()){
            channel.queueDeclare(TASK_QUEUE_NAME,false,false,false,null);
            Scanner scanner = new Scanner(System.in);
            System.out.println("请输入信息");
            while(scanner.hasNext()){
                String message = scanner.nextLine();
                channel.basicPublish("",TASK_QUEUE_NAME,null,message.getBytes());
                System.out.println("生产者task02发出消息"+ message);
            }
        }
    }
}

消费者

package com.xun.rabbitmqdemo.workQueue;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.xun.rabbitmqdemo.utils.RabbitMqUtils;
import com.xun.rabbitmqdemo.utils.SleepUtils;

public class Work03 {
    private static final String ACK_QUEUE_NAME = "ack_queue";
    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        System.out.println("Work03 等待接收消息处理时间较短");
        DeliverCallback deliverCallback = (consumerTag,delivery)->{
            String message = new String(delivery.getBody());
            SleepUtils.sleep(1);
            System.out.println("接收到消息:"+message);
            
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
        };
        CancelCallback cancelCallback = (consumerTag)->{
            System.out.println(consumerTag+"消费者取消消费接口回调逻辑");
        };
        //采用手动应答
        boolean autoAck = false;
        channel.basicConsume(ACK_QUEUE_NAME,autoAck,deliverCallback,cancelCallback);
    }
}

package com.xun.rabbitmqdemo.workQueue;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.xun.rabbitmqdemo.utils.RabbitMqUtils;
import com.xun.rabbitmqdemo.utils.SleepUtils;

public class Work04 {
    private static final String ACK_QUEUE_NAME = "ack_queue";
    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        System.out.println("Work04 等待接收消息处理时间较长");
        DeliverCallback deliverCallback = (consumerTag,delivery)->{
            String message = new String(delivery.getBody());
            SleepUtils.sleep(30);
            System.out.println("接收到消息:"+message);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
        };
        CancelCallback cancelCallback = (consumerTag)->{
            System.out.println(consumerTag+"消费者取消消费接口回调逻辑");
        };
        //采用手动应答
        boolean autoAck = false;
        channel.basicConsume(ACK_QUEUE_NAME,autoAck,deliverCallback,cancelCallback);
    }
}

工具类SleepUtils

package com.xun.rabbitmqdemo.utils;
public class SleepUtils {
    public static void sleep(int second){
        try{
            Thread.sleep(1000*second);
        }catch (InterruptedException _ignored){
            Thread.currentThread().interrupt();
        }
    }
}

模拟

work04等待30s后发出ack

在work04处理message时手动停止线程，可以看到message:dd被rabbitmq交给了work03

不公平分发

上面的轮询分发，生产者依次向消费者按顺序发送消息，但当消费者A处理速度很快，而消费者B处理速度很慢时，这种分发策略显然是不合理的。
不公平分发：

int prefetchCount = 1;
channel.basicQos(prefetchCount);

通过此配置，当消费者未处理完当前消息，rabbitmq会优先将该message分发给空闲消费者。

总结

到此这篇关于docker启动rabbitmq以及使用的文章就介绍到这了,更多相关docker启动rabbitmq及使用内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！

点击免费下载>>软考高级考试备考技巧/历年真题/备考精华资料

--结束END--

本文标题: docker启动rabbitmq以及使用方式详解

本文链接: https://www.lsjlt.com/news/171766.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

本篇文章演示代码以及资料文档资料下载

下载Word文档到电脑，方便收藏和打印～

下载Word文档

去做题

猜你喜欢

docker启动rabbitmq以及使用方式详解

目录搜索rabbitmq镜像下载镜像启动容器打印容器访问RabbitMQ Management编写生产者类消费者工作队列RabbitMqUtils工具类启动2个工作线程启动发送线程消...

99+

2022-11-13

docker运行rabbitmq rabbitmq docker 启动rabbitmq命令
Docker启动常用容器命令及配置详解

目录Docker 容器相关命令Docker 启动mysqlDocker 启动nginxDocker 启动redisDocker 启动rabbitmqDocker 启动KafkaDoc...

99+

2023-03-02

Docker启动容器命令 Docker 启动nginx Docker 启动mysql
docker安装mongoDB及使用方法详解

目录一、MongoDB是什么？1. mongo的体系结构2. mongoDB的特点(或使用场景)3. mongoDB与mysql、Redis对比4. mongoDB存储原理二、使用docker安装mongo1.安装2.创...

99+

2023-03-19

docker安装mongodb docker安装mongo
Java之MyBatis的Dao方式以及Dao动态代理详解

目录MyBatis传统Dao方式以及Dao动态代理MyBatis传统Dao方式MyBatis之Dao动态代理总结MyBatis简介以及入门参见以下文章 MyBatis简介及入门 1、...

99+

2022-12-21

Java MyBatis MyBatis的Dao方式 MyBatis Dao动态代理
SpringBoot为何可以使用Jar包启动详解

目录引言Spring Boot 打包插件SpringBoot FatJar 的组织结构MAINFEST.MF 元信息启动原理源码分析JarLauncherLauncherProper...

99+

2022-11-13
使用springboot的jar包能够以service方式启动

目录springboot jar包以service启动场景过程其他SpringBoot启动服务的三种方式1.IDEA启动2.命令行启动3.命令行编译为jar启动springboot ...

99+

2022-11-12
一文带你了解Redis怎么启动以及使用

目录一、认识Redis二、启动Redis（命令行客户端）总结一、认识Redis 特征：键值型，value支持多种不同的数据结构单线程，每个命令具备原子性（核心命令执行依然是单线程）低延迟，速度快（基于内存、IO多路复...

99+

2023-04-12

redis的启动 redis启动方法 redis启动流程
使用非root用户安装及启动docker的问题(rootless模式运行)

目录一、容器用户与宿主机用户映射二、在非root用户下运行docker守护进程2.1.docker版本要求2.2. 前置条件2.3.开始rootless模式安装2.4.启动守护进程运...

99+

2022-11-13
Vue中$attrs和$listeners详解以及使用方法

目录Vue中常见的组件通信方式可分为三类1. 前言2. 知识点3. 示例总结传送门：Vue中子组件向父组件传值及 .sync 修饰符详解传送门：Vue中状态管理...

99+

2022-11-16

vue $attrs $listeners vue $attrs $listeners属性
springboot RESTful以及参数注解的使用方式

目录springboot RESTful及参数注解使用RESTful1、@GetMapping2、@PostMapping3、@PutMapping4、@DeleteMapping5...

99+

2022-11-12
Go WaitGroup的使用方式及实例详解

WaitGroup 是 Go 语言的一个并发控制机制，它可以用于等待一组 goroutine 的结束。WaitGroup 提供了三个...

99+

2023-10-12

Go语言
Python中的模块heapq以及使用方法详解

python中的 heapq 模块 1、heapq 的两个函数：nlargest() 和 nsmallest() 1.1 nlargest(n, iterable, key=None) 函数功能：获...

99+

2023-09-01

python 开发语言队列栈 heapq
Android下拉刷新以及GridView使用方法详解

GridView是类似于ListView的控件，只是GridView可以使用多个列来呈现内容，而ListView是以行为单位，所以用法上是差不多的。主布局文件，因为要做下拉刷...

99+

2022-06-06

方法 android下拉刷新 gridview Android
ES6 Promise.all的使用方法以及其细节详解

ES6 Promise.all用法 Promise的all方法提供了并行执行异步操作的能力，并且在所有异步操作执行完后才执行回调。可以将多个 Promise 实例，包装成一个新的 P...

99+

2022-11-13
springboot 集成redission 以及分布式锁的使用详解

目录springboot集成redission及分布式锁的使用1、引入jar包2、增加Configuration类3、使用redission分布式锁Springboot整合Redis...

99+

2022-11-12
Docker镜像的迁移与备份及Dockerflie使用方法详解

目录一、迁移与备份1. 容器保存为镜像2. 镜像备份3. 镜像恢复与迁移二、Dockerflie1. 认识 Dockerfile2. Dockerfile 常用命令3. 使用 Doc...

99+

2022-11-13
详解Java CompletableFuture使用方法以及与FutureTask的区别

目录futureTask 创建异步任务创建任务1. .supplyAsync2. .runAsync异步回调1. .thenApply2. .thenAccept3. .except...

99+

2022-11-12
Docker镜像的迁移与备份及Dockerflie 使用方法详解

目录一、迁移与备份1. 容器保存为镜像2. 镜像备份3. 镜像恢复与迁移二、Dockerflie1. 认识 Dockerfile2. Dockerfile 常用命令3. 使用 Dockerfile 构建镜像一、迁移与备份迁移和备份主要有...

99+

2022-09-23
vue动态绑定ref(使用变量)以及获取方式

目录vue动态绑定ref及获取一起来看下代码吧ref三种使用方法refvue动态绑定ref及获取正常情况，我们需要在vue中获得某个dom或者组件，我们会通过绑定 ref 然后通过...

99+

2022-11-13
ASP.NET Core使用HostingStartup增强启动操作方法详解

概念在ASP.NET Core中我们可以使用一种机制来增强启动时的操作，它就是HostingStartup。如何叫"增强"操作，相信了解过AOP概念的同学应该都非常的熟悉。我...

99+

2022-06-07

net ASP.NET 方法启动 core ASP