原 Mac下RabbitMq安装与测试教程

RabbitMq基础教程之安装与测试

Installing on Mac

I. 安装

brew install rabbitmq

## 进入安装目录
cd /usr/local/Cellar/rabbitmq/3.7.5

# 启动
brew services start rabbitmq
# 当前窗口启动
rabbitmq-server

<!-- more -->

启动控制台之前需要先开启插件

./rabbitmq-plugins enable rabbitmq_management

进入控制台: http://localhost:15672/

用户名和密码:guest,guest

II. 配置与测试

1. 添加账号

首先是得启动mq

## 添加账号
./rabbitmqctl add_user admin admin
## 添加访问权限
./rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"
## 设置超级权限
./rabbitmqctl set_user_tags admin administrator

2. 编码实测

pom引入依赖

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
</dependency>

开始写代码

public class RabbitMqTest {

    //消息队列名称
    private final static String QUEUE_NAME = "hello";

    @Test
    public void send() throws java.io.IOException, TimeoutException {

        //创建连接工程
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        //创建连接
        Connection connection = factory.newConnection();

        //创建消息通道
        Channel channel = connection.createChannel();

        //生成一个消息队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);


        for (int i = 0; i < 10; i++) {
            String message = "Hello World RabbitMQ count: " + i;

            //发布消息,第一个参数表示路由(Exchange名称),未""则表示使用默认消息路由
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());

            System.out.println(" [x] Sent '" + message + "'");
        }


        //关闭消息通道和连接
        channel.close();
        connection.close();

    }


    @Test
    public void consumer() throws java.io.IOException, java.lang.InterruptedException, TimeoutException {

        //创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");

        //创建连接
        Connection connection = factory.newConnection();

        //创建消息信道
        Channel channel = connection.createChannel();

        //消息队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        System.out.println("[*] Waiting for message. To exist press CTRL+C");

        AtomicInteger count = new AtomicInteger(0);

        //消费者用于获取消息信道绑定的消息队列中的信息
        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");

                try {
                    System.out.println(" [x] Received '" + message);
                } finally {
                    System.out.println(" [x] Done");
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        channel.basicConsume(QUEUE_NAME, false, consumer);

        Thread.sleep(1000 * 60);
    }
}

需要注意的一点是:

  • 生产消息: channel.queueDeclare(QUEUE_NAME, true, false, false, null);
  • 消费消息: channel.queueDeclare(QUEUE_NAME, true, false, false, null);
  • 生产和消费都声明channel,要求两者的配置参数一致,否则无法消费数据

3. 输出说明

首先执行塞入数据,执行完毕之后,可以到控制台进行查看:

可以看到多出了一个Queue,对列名为hello,总共有10条数据


接下来就是消费数据了,执行consumer方法,输出日志

[*] Waiting for message. To exist press CTRL+C
 [x] Received 'Hello World RabbitMQ count: 0
 [x] Done
 [x] Received 'Hello World RabbitMQ count: 1
 [x] Done
 [x] Received 'Hello World RabbitMQ count: 2
 [x] Done
 [x] Received 'Hello World RabbitMQ count: 3
 [x] Done
 [x] Received 'Hello World RabbitMQ count: 4
 [x] Done
 [x] Received 'Hello World RabbitMQ count: 5
 [x] Done
 [x] Received 'Hello World RabbitMQ count: 6
 [x] Done
 [x] Received 'Hello World RabbitMQ count: 7
 [x] Done
 [x] Received 'Hello World RabbitMQ count: 8
 [x] Done
 [x] Received 'Hello World RabbitMQ count: 9
 [x] Done

回头去查看queue,发现总得数据量为0了

4. ACK问题

对于ack的问题,如果在消费数据的时候,出现异常,而我不希望数据丢失,这个时候就需要考虑手动ack的机制来保证了

首先需要设置手动ack

// 设置autoAck为false
channel.basicConsume(QUEUE_NAME, false, consumer);

其次在消费数据完毕之后,主动ack/nack

if (success) {
    channel.basicAck(envelope.getDeliveryTag(), false);
} else {
    channel.basicNack(envelope.getDeliveryTag(), false, false);
}

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏张善友的专栏

LINQ via C# 系列文章

LINQ via C# Recently I am giving a series of talk on LINQ. the name “LINQ via C...

2605
来自专栏java 成神之路

使用 NIO 实现 echo 服务器

4507
来自专栏张善友的专栏

Silverlight + Model-View-ViewModel (MVVM)

     早在2005年,John Gossman写了一篇关于Model-View-ViewModel模式的博文,这种模式被他所在的微软的项目组用来创建Expr...

2898
来自专栏我和未来有约会

Kit 3D 更新

Kit3D is a 3D graphics engine written for Microsoft Silverlight. Kit3D was inita...

2506
来自专栏pangguoming

Spring Boot集成JasperReports生成PDF文档

由于工作需要,要实现后端根据模板动态填充数据生成PDF文档,通过技术选型,使用Ireport5.6来设计模板,结合JasperReports5.6工具库来调用渲...

1.2K7
来自专栏转载gongluck的CSDN博客

cocos2dx 打灰机

#include "GamePlane.h" #include "PlaneSprite.h" #include "BulletNode.h" #include...

5286
来自专栏一个爱瞎折腾的程序猿

sqlserver使用存储过程跟踪SQL

USE [master] GO /****** Object: StoredProcedure [dbo].[sp_perfworkload_trace_s...

1990
来自专栏Ceph对象存储方案

Luminous版本PG 分布调优

Luminous版本开始新增的balancer模块在PG分布优化方面效果非常明显,操作也非常简便,强烈推荐各位在集群上线之前进行这一操作,能够极大的提升整个集群...

3035
来自专栏一个会写诗的程序员的博客

Spring Reactor 项目核心库Reactor Core

Non-Blocking Reactive Streams Foundation for the JVM both implementing a Reactiv...

2102
来自专栏闻道于事

js登录滑动验证,不滑动无法登陆

js的判断这里是根据滑块的位置进行判断,应该是用一个flag判断 <%@ page language="java" contentType="text/html...

6588

扫码关注云+社区