安裝Erlang

RabbitMQ基于Erlang洞豁，所以得先安裝Erlang

http://www.erlang.org/downloads

根據(jù)自己的系統(tǒng)選擇下載寝衫，安裝完了瞧哟，配置一下path即可

windows默認(rèn)安裝路徑：C:\Program Files\erl9.3\bin;

驗(yàn)證： erl -version

有時(shí)候可能需要重啟才能生效

安裝RabbitMQ

http://www.rabbitmq.com/install-windows.html

默認(rèn)安裝路徑最好修改一下且轨，因?yàn)镽abbitMQ不支持帶有空格的路徑（需先安裝Erlang）

安裝RabbitMQ-Plugins

這個(gè)是管理界面毛肋，可以查看隊(duì)列消息及各種信息

• 進(jìn)入rabbitmq的sbin目錄
• 輸入rabbitmq-plugins enable rabbitmq_management命令(需server已啟動(dòng) rabbitmq-service start)
• 驗(yàn)證 http://localhost:15672
• 用戶名密碼都是guest

rabbitmq_01.png

RabbitMQ的簡(jiǎn)單介紹

先看幾個(gè)概念

• producer：生產(chǎn)者

• consumer：消費(fèi)者

• virtual host：虛擬主機(jī)

  • 在RabbitMQ中梨水，用戶只能在虛擬主機(jī)的層面上進(jìn)行一些權(quán)限設(shè)置，比如可以訪問哪些隊(duì)列僚饭，可以處理哪些請(qǐng)求等

• broker：消息轉(zhuǎn)發(fā)者

  • 也就是我們RabbitMQ服務(wù)端充當(dāng)?shù)墓δ?/li>
  • exchange：交換機(jī)
    • 和producer直接打交道的震叮，主要進(jìn)行轉(zhuǎn)發(fā)操作
  • queue：消息隊(duì)列
    • 用于接收exchange路由過來的消息并存放

send

package com.jiataoyuan.demo.rabbitmq.config;

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

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author TaoYuan
 * @version V1.0.0
 * @date 2018/4/21 0021
 * @description producer 生產(chǎn)者
 *
 * (1)：創(chuàng)建ConnectionFactory，并且設(shè)置一些參數(shù)鳍鸵，比如hostname,portNumber等等
 * (2)：利用ConnectionFactory創(chuàng)建一個(gè)Connection連接
 * (3)：利用Connection創(chuàng)建一個(gè)Channel通道
 * (4)：創(chuàng)建queue并且和Channel進(jìn)行綁定
 * (5)：創(chuàng)建消息冤荆，并且發(fā)送到隊(duì)列中
 *
 * 本例沒有用到exchange交換機(jī)，RabbitMQ默認(rèn)情況下是會(huì)創(chuàng)建一個(gè)空字符串名字的exchange
 * 如果我們沒有創(chuàng)建自己的exchange的話权纤，默認(rèn)就是使用的這個(gè)exchange
 */
public class Send {
    private final static String QUEUE_NAME = "MyQueue";

    public static void main(String[] args) {
        send();
    }

    public static void send()
    {
        ConnectionFactory factory = null;
        Connection connection = null;
        Channel channel = null;
        try {
            factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            connection = factory.newConnection();
            channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            String message = "Send MyQueue send message .....";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
            System.out.println("已經(jīng)發(fā)送消息....."+message);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }finally{
            try {
                //關(guān)閉資源
                channel.close();
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
        }
    }
}

receive

package com.jiataoyuan.demo.rabbitmq.config;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author TaoYuan
 * @version V1.0.0
 * @date 2018/4/21 0021
 * @description consumer 消費(fèi)者
 *
 * (1)：創(chuàng)建ConnectionFactory，并且設(shè)置一些參數(shù)乌妒，比如hostname,portNumber等等
 * (2)：利用ConnectionFactory創(chuàng)建一個(gè)Connection連接
 * (3)：利用Connection創(chuàng)建一個(gè)Channel通道
 * (4)：將queue和Channel進(jìn)行綁定汹想，注意這里的queue名字要和前面producer創(chuàng)建的queue一致
 * (5)：創(chuàng)建消費(fèi)者Consumer來接收消息，同時(shí)將消費(fèi)者和queue進(jìn)行綁定
 *
 */
public class Receive {
    private final static String QUEUE_NAME = "MyQueue";

    public static void main(String[] args) {
        receive();
    }

    public static void receive()
    {
        ConnectionFactory factory = null;
        Connection connection = null;
        Channel channel = null;

        try {
            factory = new ConnectionFactory();
            factory.setHost("localhost");
            connection = factory.newConnection();
            channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);

            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("收到消息....."+message);
                }
            };


            channel.basicConsume(QUEUE_NAME, true,consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }finally{
            try {
                //關(guān)閉資源
                channel.close();
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
        }
    }
}

做測(cè)試的時(shí)候撤蚊，可以先發(fā)送古掏，不接收，然后去 http://localhost:15672/#/queues 看看侦啸。

SpringBoot整合RabbitMQ

以上就是RabbitMQ的基本用法槽唾，接下來還是要整合到SpringBoot中使用丧枪。

依賴

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

配置

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.virtual-host=/

RabbitMQ模式有很多，還是演示一下最簡(jiǎn)單的模式庞萍，實(shí)際開發(fā)過程中可以根據(jù)業(yè)務(wù)選擇最適合的業(yè)務(wù)場(chǎng)景

Sender

package com.jiataoyuan.demo.rabbitmq.config;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * Created by Administrator on 2017/5/8 0008.
 */
@Component
public class Sender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendData(String data){
        if (null == data){
            data = "data is null! Time: " + new Date();
        }
        System.out.println("Sender : " + data);
        rabbitTemplate.convertAndSend("hello", data);
    }

}

Receive

package com.jiataoyuan.demo.rabbitmq.config;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @author TaoYuan
 * @version V1.0.0
 * @date 2018/4/21 0021
 * @description description
 */
@Component
@RabbitListener(queues = "hello")
public class Receive {

    @RabbitHandler
    public void process(String hello) {
        System.out.println("Receiver  : " + hello);
    }
}

Controller

package com.jiataoyuan.demo.rabbitmq.controller;

import com.jiataoyuan.demo.rabbitmq.config.Sender;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.Date;

/**
 * @author TaoYuan
 * @version V1.0.0
 * @date 2018/4/21 0021
 * @description description
 */
@RestController
@RequestMapping("/rabbit")
public class RabbitMQController {

    @Resource
    private Sender sender;


    @GetMapping()
    public String Main(){
        return "<h1>hello RabbitMQ!</h1>";
    }

    @GetMapping("/send")
    public String Send() throws Exception{
        sender.sendData("Hello, This is OneToOne!");
        return "Send OK拧烦！";
    }

}

result

Sender : Hello, This is OneToOne!
Receiver  : Hello, This is OneToOne!