Kafka、RocketMQ、RabbitMQ的优劣势比较

最全MQ消息队列有哪些？ 目前在业界有哪些比较知名的消息引擎呢？如下图所示 这里面几乎完全列举了当下比较知名的消息引擎，包括： ZeroMQ 推特的Distributedlog ActiveMQ：Apache旗下的老牌消息引擎 RabbitMQ、Kafka：AMQP的默认实现。 RocketMQ Artemis：Apache的ActiveMQ下的子项目 Apollo：同样为Apache的ActiveMQ的子项目的号称下一代消息引擎 商业化的消息引擎IronMQ 以及实现了JMS(Java Message Service)标准的OpenMQ。   MQ消息队列的技术应用 1.解耦 解耦是消息队列要解决的最本质问题。 2.最终一致性 最终一致性指的是两个系统的状态保持一致，要么都成功，要么都失败。 最终一致性不是消息队列的必备特性，但确实可以依靠消息队列来做最终一致性的事情。 2.广播 消息队列的基本功能之一是进行广播。 有了消息队列，我们只需要关心消息是否送达了队列，至于谁希望订阅，是下游的事情，无疑极大地减少了开发和联调的工作量。 3.错峰与流控 典型的使用场景就是秒杀业务用 […]

Kafka

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。 这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。

rabbitmq

RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。RabbitMQ服务器是用Erlang语言编写的，而集群和故障转移是构建在开放电信平台框架上的。所有主要的编程语言均有与代理接口通讯的客户端库。

RabbitMQ消息持久化

目录 一、前言 二、队列持久化 　　2.1 查看存在的队列和消息数量 　　2.2 持久化队列 三、消息持久化 四、总结 一、前言 如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。当然还是会有一些小概率事件会导致消息丢失。 二、队列持久化 　　2.1 查看存在的队列和消息数量 在windows环境下，在rabbitmq的安装目录/sbin下，通过rabbitmqctl.bat list_queues查看　 这边启动了两个producer,分别生成两个队列hello 和 hello1,并且他们都有一个消息存在 重启rabbitmq，模拟故障 可以看到重启后两个队列都消失了 　　2.2 持久化队列 我们就hello队列持久化 在声明队列名称时，持久化队列，生产端和消费端都要 1 channel.queue_declare(queue=’hello', durable=True) 我们重复上面的操作，但是给hello队列做持久化，而hell […]

用rabbitMq解决web高并发的学习笔记

引入pom.xml

1
2
3
4
5
6
7
<!-- https://docs.spring.io/spring-amqp/docs/2.0.2.RELEASE/reference/html/
        https://docs.spring.io/spring-boot/docs/current/reference/html/boot-features-messaging.html#boot-features-amqp
         -->
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>

  application.yml 配置部分

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
    #rabbitmq
    rabbitmq:
      host: 10.0.0.2
      port: 5672
      username: springboot
      password: password
      publisher-confirms: true
      publisher-returns: true
      template:
         mandatory: true
      #https://github.com/spring-projects/spring-boot/blob/v2.0.5.RELEASE/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitProperties.java
      listener:
         concurrency: 2
         #最小消息监听线程数
         max-concurrency: 2
         #最大消息监听线程数
mybatis:
  mapper-locations: classpath:mapping/*.xml
···

  创建配置文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
package com.redis;
 
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
//错误的 import com.rabbitmq.client.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
 
/**
*
* Broker:它提供一种传输服务,它的角色就是维护一条从生产者到消费者的路线，保证数据能按照指定的方式进行传输,
* Exchange：消息交换机,它指定消息按什么规则,路由到哪个队列。 Queue:消息的载体,每个消息都会被投到一个或多个队列。
*
* Binding:绑定，它的作用就是把exchange和queue按照路由规则绑定起来.
*
* Routing Key:路由关键字,
*
* exchange根据这个关键字进行消息投递。 vhost:虚拟主机,一个broker里可以有多个vhost，用作不同用户的权限分离。
*
* Producer:消息生产者,就是投递消息的程序. Consumer:消息消费者,就是接受消息的程序.
* Channel:消息通道,在客户端的每个连接里,可建立多个channel.
*
*
*/
 
@Configuration
public class RabbitConfig {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
    @Value("${spring.rabbitmq.host}")
    private String host;
 
    @Value("${spring.rabbitmq.port}")
    private int port;
 
    @Value("${spring.rabbitmq.username}")
    private String username;
 
    @Value("${spring.rabbitmq.password}")
    private String password;
 
    public static final String EXCHANGE_A = "my-mq-exchange_A";
    
    public static final String QUEUE_A = "QUEUE_A";
    
    public static final String ROUTINGKEY_A = "spring-boot-routingKey_A";
 
 
    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setPublisherConfirms(true);
        
        
        return connectionFactory;
    }
 
    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    // 必须是prototype类型
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        return template;
    }
 
    /**
     * * 针对消费者配置 *
     * 1. 设置交换机类型 *
     * 2. 将队列绑定到交换机
     *
     * FanoutExchange:
     * 将消息分发到所有的绑定队列，无routingkey的概念
     *
     * HeadersExchange ：通过添加属性key-value匹配
     * DirectExchange:按照routingkey分发到指定队列
     * TopicExchange:多关键字匹配
     *
     * 如果需要使用的其他的交换器类型，spring中都已提供实现，所有的交换器均实现org.springframework.amqp.core.AbstractExchange接口。
常用交换器类型如下：
 
        Direct(DirectExchange)：direct 类型的行为是"先匹配, 再投送".
        即在绑定时设定一个 routing_key, 消息的routing_key完全匹配时, 才会被交换器投送到绑定的队列中去。
        
        Topic(TopicExchange)：按规则转发消息（最灵活）。
        
        Headers(HeadersExchange)：设置header attribute参数类型的交换机。
        
        Fanout(FanoutExchange)：转发消息到所有绑定队列。
     *
     */
    
    
 
 
    
 
    
    
    @Bean
    public DirectExchange defaultExchange() {
        return new DirectExchange(EXCHANGE_A);
    }
 
    /**
     * 获取队列A
     *
     * @return
     */
    @Bean
    public Queue queueA() {
        return new Queue(QUEUE_A, true); // 队列持久
    }
 
 
 
    /**
     * 一个交换机可以绑定多个消息队列，也就是消息通过一个交换机，可以分发到不同的队列当中去。
     *
     * @return
     */
    @Bean
    public Binding binding() {
 
        return BindingBuilder.bind(queueA()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_A);
    }
 
    
 
}

  Producer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
package com.redis;
 
import java.util.UUID;
import org.springframework.amqp.core.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
 
@Component
public class MsgProducer implements RabbitTemplate.ConfirmCallback , ReturnCallback{
    
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    
    // 由于rabbitTemplate的scope属性设置为ConfigurableBeanFactory.SCOPE_PROTOTYPE，所以不能自动注入
    private RabbitTemplate rabbitTemplate;
 
    /** * 构造方法注入rabbitTemplate */
    @Autowired
    public MsgProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        rabbitTemplate.setConfirmCallback(this);
        // rabbitTemplate如果为单例的话，那回调就是最后设置的内容
        
        /**
         * ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调。
         * ReturnCallback接口用于实现消息发送到RabbitMQ交换器，但无相应队列与交换器绑定时的回调。
         */
        
        rabbitTemplate.setReturnCallback(this);
    }
    
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.out.println(message.getMessageProperties().getCorrelationIdString() + " 发送失败");
        
        System.out.println("消息主体 message : "+message);
        System.out.println("消息主体 message : "+replyCode);
        System.out.println("描述："+replyText);
        System.out.println("消息使用的交换器 exchange : "+exchange);
        System.out.println("消息使用的路由键 routing : "+routingKey);
        
    }
    
     /**
      * rabbitTemplate.send(message);   //发消息，参数类型为org.springframework.amqp.core.Message
        rabbitTemplate.convertAndSend(object); //转换并发送消息。 将参数对象转换为org.springframework.amqp.core.Message后发送
        rabbitTemplate.convertSendAndReceive(message) //转换并发送消息,且等待消息者返回响应消息。
    
      * @param content
      */
 
    public void sendMsg(String content) {
        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
        
        System.out.println("开始发送消息c : " + content.toLowerCase() + " ,correlationId= " + correlationId);
        String response = rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_A, RabbitConfig.ROUTINGKEY_A, content, correlationId).toString();
        System.out.println("结束发送消息c : " + content.toLowerCase());
        System.out.println("消费者响应c : " + response + " 消息处理完成");
        //logger.info(" 发送消息TO A:" + content);
        // 把消息放入ROUTINGKEY_A对应的队列当中去，对应的是队列A
        //rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_A, RabbitConfig.ROUTINGKEY_A, content, correlationId);
    }
 
    /** * 回调 */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        logger.info(" 回调id:" + correlationData);
        if (ack) {
            logger.info("消息成功消费");
        } else {
            logger.info("消息消费失败:" + cause);
        }
        
        
    
    }
 
}

  Receiver

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
package com.redis;
 
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
@Component
@RabbitListener(queues = RabbitConfig.QUEUE_A)
public class MsgReceiver {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
    @RabbitHandler
    public void process(String content) {
        System.out.println("接收处理队列A当中的消息： " + content);
    }
}

  unit test

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import java.util.Date;
 
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
 
import com.redis.MsgProducer;
 
@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMqTest {
 
    @Autowired
    private MsgProducer sender;
    
    @Test
    public void sendTest() throws Exception {
        //while(true){
            String msg = new Date().toString();
            sender.sendMsg(msg);
            Thread.sleep(6000);
        //}
    }
 
}

  可以测试通过 image.png 高并发的访问量， 除了使用nginx/haproxy本身实现外， 尝试了google guava 简单好用， 来控制前端用户请求量， 范例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
package com.test.ratelimit;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
 
import com.google.common.util.concurrent.RateLimiter;
 
 
public class ComplexDemo {
    
    private static RateLimiter rateLimiter = RateLimiter.create(10);
    
    private static AtomicInteger suc = new AtomicInteger(0), fail = new AtomicInteger(0);
 
    public static void main(String[] args) {
        // TODO Auto-generated method stub
        
        List<Runnable> tasks = new ArrayList<Runnable>();  
        for (int i = 0; i < 100; i++) {  
            tasks.add(new UserRequest(i));  
        }  
        ExecutorService threadPool = Executors.newCachedThreadPool();  
        for (Runnable runnable : tasks) {
            threadPool.execute(runnable);  
        }
        
        
    }
    
    
    private static boolean startGo(int i) {
         //基于令牌桶算法的限流实现类
        
        /**
         * 一秒出10个令牌，0.1秒出一个，100个请求进来，假如100个是同时到达， 那么最终只能成交10个，90个都会因为超时而失败。
         *      
         */
        
        /**
         * tryAcquire(long timeout, TimeUnit unit)
         * 从RateLimiter 获取许可如果该许可可以在不超过timeout的时间内获取得到的话，
         * 或者如果无法在timeout 过期之前获取得到许可的话，那么立即返回false（无需等待）
         */  
        
        //判断能否在1秒内得到令牌，如果不能则立即返回false，不会阻塞程序  
        if (!rateLimiter.tryAcquire(1000, TimeUnit.MILLISECONDS)) {  
            System.out.println("暂时无法获取令牌， 排队失败" + i);  
            fail.getAndIncrement();
            System.out.println("SUC/FAIL=" + suc.get() + "/" + fail.get());  
            return false;
        }  
        if (update() > 0) {  
            System.out.println("成功" + i);  
            suc.getAndIncrement();
            System.out.println("FAIL/SUC=" + fail.get() + "/" + suc.get());  
            return true;
        }  
        
        System.out.println("数据不足，失败");  
        return false;
    }
    
    
    private static int update() {
        return 1;
    }
    
     private static class UserRequest implements Runnable {  
            private int id;  
      
            public UserRequest(int id) {  
                this.id = id;  
            }  
      
            public void run() {  
                startGo(id) ;
            }  
        }  
 
}
 
测试结果：
...
成功8
FAIL/SUC=89/10
成功7
FAIL/SUC=89/11

  总结， rabbit mq, 这里只用Direct。 Topic匹配灵活， 可以用到其他场景。 REF: http://www.rab […]

服务级高并发秒杀优化（RabbitMQ+接口优化）

1. 安装RabbitMQ linux下的安装没什么可说的，因为本机懒得重装虚拟机了，所以就下载了windows版本进行安装。 erlang下载地址：http://www.erlang.org/download.html rabbitMQ下载：http://www.rabbitmq.com/download.html 直接下载安装即可。 因为想用可视化界面监控消息，所以先激活这个功能。

1
2
//到rabbitMQ安装目录的sbin目录下启动cmd黑窗口
E:\software\RabbitMQServer\rabbitmq_server-3.6.5\sbin>rabbitmq-plugins.bat enable rabbitmq_management

  然后重启rabbitMQ服务。输入网址：http://localhost:15672/。 使用默认用户guest/guest进入网页端控制台。 2. rabbitMQ基本原理和使用 rabbitMQ原理 Broker：简单来说就是消息队列服务器实体。 Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。 Queue：消息队列载体，每个消息都会被投入到一个或多个队列。 Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。 Routing Key：路由关键字，exchange根据 […]

.Net高并发解决思路

首先在windows上安装好Redis，RabbitMQ Redis-cli使用示例 ModelContext.cs代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
    public class ModelContext : DbContext
    {
        //您的上下文已配置为从您的应用程序的配置文件(App.config 或 Web.config)
        //连接字符串。
        public ModelContext()
            : base("name=default")
        {
        }
 
        public virtual DbSet<Person> Person { get; set; }
    }
 
    public class Person
    {
        public int Id { get; set; }
        public string Id2 { get; set; }
        public string Name { get; set; }
    }

在 Package Manager Console 下运行命令 Enable-Migrations 这个命令将在项目下创建文件夹 Migrations The Configuration class 这个类允许你去配置如何迁移，对于本文将使用默认的配置（在本文中因为只有一个 Context, Enable-Migrations 将自动对 context type 作出适配）; An InitialCreate migration （本文为201702220232375_20170222.cs）这个迁移之所以存在是因为我们之前用 Code First 创建了数据库, 在启用迁移前，scaffolded migration 里的代码表示在数据库中已经创建的对象，本文中即为表 Person（列 Id 和 Name）. 文件名包含一个 timestamp 以便排序（如果之前数据库没有被创建，那么 Initi […]

RabbitMQ的几种典型使用场景

AMQP AMQP协议是一个高级抽象层消息通信协议，RabbitMQ是AMQP协议的实现。它主要包括以下组件： 1.Server(broker): 接受客户端连接，实现AMQP消息队列和路由功能的进程。 2.Virtual Host:其实是一个虚拟概念，类似于权限控制组，一个Virtual Host里面可以有若干个Exchange和Queue，但是权限控制的最小粒度是Virtual Host 3.Exchange:接受生产者发送的消息，并根据Binding规则将消息路由给服务器中的队列。ExchangeType决定了Exchange路由消息的行为，例如，在RabbitMQ中，ExchangeType有direct、Fanout和Topic三种，不同类型的Exchange路由的行为是不一样的。 4.Message Queue：消息队列，用于存储还未被消费者消费的消息。 5.Message: 由Header和Body组成，Header是由生产者添加的各种属性的集合，包括Message是否被持久化、由哪个Message Queue接受、优先级是多少等。而Body是真正需要传输的APP数据。 […]

高并发场景之RabbitMQ篇

上次我们介绍了在单机、集群下高并发场景可以选择的一些方案，传送门：高并发场景之一般解决方案 但是也发现了一些问题，比如集群下使用ConcurrentQueue或加锁都不能解决问题，后来采用Redis队列也不能完全解决问题， 因为使用Redis要自己实现分布式锁   这次我们来了解一下一个专门处理队列的组件：RabbitMQ，这个东西天生支持分布式队列。 下面我们来用RabbitMQ来实现上一篇的场景   一、新建RabbitMQ.Receive

1
2
private static ConnectionFactory factory = new ConnectionFactory
{ HostName = "192.168.1.109", UserName = "ljr", Password = "root", VirtualHost = "/" };

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
1         static void Main(string[] args)
2         {
3             using (var connection = factory.CreateConnection())
4             {
5                 using (var channel = connection.CreateModel())
6                 {
7                     var consumer = new EventingBasicConsumer();
8                     consumer.Received += (model, ea) =&gt;
9                     {
10                         var body = ea.Body;
11                         var message = Encoding.UTF8.GetString(body);
12                         Console.WriteLine(" [x] Received {0}", message);
13
14                         var total = DbHelper.ExecuteScalar("Select Total from ConCurrency where Id = 1", null).ToString();
15                         var value = int.Parse(total) + 1;
16
17                         DbHelper.ExecuteNonQuery(string.Format("Update ConCurrency Set Total = {0} where Id = 1", value.ToString()), null);
18                     };
19
20                     channel.QueueDeclare(queue: "queueName", durable: false, exclusive: false, autoDelete: false, arguments: null);
21                     channel.BasicConsume(queue: "queueName", noAck: true, consumer: consumer);
22
23                     Console.WriteLine(" Press [enter] to exit.");
24                     Console.ReadLine();
25                 }
26             }
27         }

二、新建RabbitMQ.Send

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
1         static void Main(string[] args)
2         {
3             for (int i = 1; i &lt;= 500; i++)
4             {
5                 Task.Run(async () =&gt;
6                 {
7                     await Produce();
8                 });
9
10                 Console.WriteLine(i);
11             }
12
13             Console.ReadKey();
14         }
15
16         public static Task Produce()
17         {
18             return Task.Factory.StartNew(() =&gt;
19             {
20                 using (var connection = factory.CreateConnection())
21                 {
22                     using (var channel = connection.CreateModel())
23                     {
24                         var body = Encoding.UTF8.GetBytes(Guid.NewGuid().ToString());
25                         channel.QueueDeclare(queue: "queueName", durable: false, exclusive: false, autoDelete: false, arguments: null);
26                         channel.BasicPublish(exchange: "", routingKey: "queueName", basicProperties: null, body: body);
27                     }
28                 }
29             });
30         }

  这里是模拟500个用户请求，正常的话最后Total就等于500 我们来说试试看，运行程序 2.1、打开接收端 2.2 运行客户端 2.3、可以看到2边几乎是实时的，再去看看数据库 三、我们在集群里执行   最后数据是1000   完全没有冲突，好了，就是这样 。 […]

在CentOS7上安装RabbitMQ

安装过程参考官网： Installing on RPM-based Linux (RHEL, CentOS, Fedora, openSUSE) 首先需要安装erlang，参考：http://fedoraproject.org/wiki/EPEL/FAQ#howtouse

1
2
rpm -Uvh https://mirrors.tuna.tsinghua.edu.cn/epel/epel-release-latest-7.noarch.rpm
yum install erlang

安装过程中会有提示，一路输入“y”即可。 完成后安装RabbitMQ： 先下载rpm：

1
wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.6/rabbitmq-server-3.6.6-1.el7.noarch.rpm

下载完成后安装：

1
yum install rabbitmq-server-3.6.6-1.el7.noarch.rpm

完成后启动服务：

1
service rabbitmq-server start

可以查看服务状态：

1
service rabbitmq-server status

这里可以看到log文件的位置，转到文件位置，打开文件： 这里显示的是没有找到配置文件，我们可以自己创建这个文件

1
2
cd /etc/rabbitmq/
vi rabbitmq.config

编辑内容如下： [crayon-6107de9115b7b57473294 […]

RabbitMQ用户增删及权限控制

用户角色分类 none：无法登录控制台 不能访问 management plugin，通常就是普通的生产者和消费者。 management：普通管理者。 仅可登陆管理控制台(启用management plugin的情况下)，无法看到节点信息，也无法对policies进行管理。用户可以通过AMQP做的任何事外加： 列出自己可以通过AMQP登入的virtual hosts 查看自己的virtual hosts中的queues, exchanges 和 bindings 查看和关闭自己的channels 和 connections 查看有关自己的virtual hosts的“全局”的统计信息，包含其他用户在这些virtual hosts中的活动。 policymaker：策略制定者。 management可以做的任何事外加： 查看、创建和删除自己的virtual hosts所属的policies和parameters monitoring：监控者。 management可以做的任何事外加： 列出所有virtual hosts，包括他们不能登录的virtual hosts 查看其他用户的con […]

RabbitMQ从入门到精通—安装

安装脚本

1
2
3
4
5
6
7
8
9
10
yum install wget -y
 
#Download RabbitMQ and Erlang
wget https://www.rabbitmq.com/releases/rabbitmq-server/v3.6.9/rabbitmq-server-3.6.9-1.el6.noarch.rpm
wget https://www.rabbitmq.com/releases/erlang/erlang-19.0.4-1.el6.x86_64.rpm
 
#Install Erlang and RabbitMQ
rpm -ivh erlang-19.0.4-1.el6.x86_64.rpm
rpm --import http://www.rabbitmq.com/rabbitmq-signing-key-public.asc
yum install rabbitmq-server-3.6.9-1.el6.noarch.rpm -y

发现错误

1
2
1. 安装依赖Requires: socat
2. 安装socat的时候提示需要tcp_wrappers

解决方案

1
2
3
4
wget –no-cache http://www.convirture.com/repos/definitions/rhel/6.x/convirt.repo -O /etc/yum.repos.d/convirt.repo
yum install socat -y
 
使用https://rpmfind.net/linux/rpm2html/search.php搜索tcp_wrappers，下载，rpm -ivh tcp_wrappers.version

启动RabbitMQ

1
2
3
4
5
6
chkconfig rabbitmq-server on
/sbin/service rabbitmq-server start
# 修改密码
rabbitmqctl change_password guest welcome123
# 启动管理界面
rabbitmq-plugins enable rabbitmq_management

使用过程错误小记 user ‘guest’ – User can only log in via localhost 登录响应如下：

1
{"error":"not_authorised","reason":"User can only log in via localhost"}

我实在docker中创建rabbitmq的，然后将端口15672映射到宿主机器上的15672，所以也算是远程登录，而guest用户要求只能使用localhost登录，所以要想远程登录必须新建用户。

1
2
3
rabbitmqctl add_user {username} {password}
# 例：创建admin用户
rabbitmqctl add_user admin 123456

新建用户无法登录

1
{"error":"not_authorised","reason":"Not management user"}

需要授予用户角色

1
2
3
rabbitmqctl set_user_tags {username} {role}
# 例
rabbitmqctl set_user_tags admin administrator

用户授权 [crayon-6107de9 […]