1.安裝

下載

erlang-19.0.4-1.el7.centos.x86_64.rpm
https://www.rabbitmq.com/releases/erlang/
http://erlang.org/download/
http://repo.iotti.biz/CentOS/7/x86_64/socat-1.7.3.2-5.el7.lux.x86_64.rpm
https://www.rabbitmq.com/install-rpm.html#downloads
注意erlang版本對應(yīng)rabbitmq版本

1.erlang安裝
curl -s https://packagecloud.io/install/repositories/rabbitmq/erlang/script.rpm.sh | sudo bash
yum install -y erlang
elk
2.socat安裝
wget http://repo.iotti.biz/CentOS/7/x86_64/socat-1.7.3.2-5.el7.lux.x86_64.rpm
rpm -ivh socat-1.7.3.2-5.el7.lux.x86_64.rpm
3.rabbitmq
wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.8.10/rabbitmq-server-3.8.10-1.el8.noarch.rpm
rpm -ivh rabbitmq-server-3.8.10-1.el7.noarch.rpm
4.開啟插件
rabbitmq-plugins enable rabbitmq_management
5.重啟
systemctl restart rabbitmq-server.service
6.查看運(yùn)行狀態(tài)
systemctl status rabbitmq-server.service

7.控制臺(tái)訪問
http://192.168.0.104:15672/
guest/guest 
8.報(bào)錯(cuò)error:User can only log in via localhost
添加用戶
rabbitmqctl add_user admin admin
rabbitmqctl set_user_tags admin administrator
rabbitmqctl set_permissions -p / admin "." "." ".*"
systemctl restart rabbitmq-server
使用admin/admin登陸控制臺(tái)
安裝ok

2.java代碼demo

演示第一種

image.png

2.1 生產(chǎn)者

public class ConnUtils {

    public static ConnectionFactory getConnection() {
        ConnectionFactory con = new ConnectionFactory();
        con.setPort(5672);
        con.setHost("192.168.43.84");
        con.setUsername("ems");
        con.setPassword("ems");
        con.setVirtualHost("/ems");
        return con;
    }
}


public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory con = ConnUtils.getConnection();

        // 創(chuàng)建連接對象
        Connection connection = con.newConnection();
        // 創(chuàng)建通道
        Channel channel = connection.createChannel();

        /**
         * 1. 隊(duì)列名稱
         * 2.是否持久化
         * 3.是否獨(dú)占隊(duì)列
         * 4.是否消費(fèi)完后自動(dòng)刪除隊(duì)列
         * 5.額外參數(shù)
         */
        channel.queueDeclare("hello", false, false, false, null);

        // 1.交換機(jī) 2.隊(duì)列名 3.傳遞消息額外設(shè)置 4.消息具體內(nèi)容
        channel.basicPublish("","hello", MessageProperties.PERSISTENT_TEXT_PLAIN,
            "hello word".getBytes(StandardCharsets.UTF_8));

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

image.png

2.3.消費(fèi)者

public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory con = ConnUtils.getConnection();
        // 創(chuàng)建連接對象
        Connection connection = con.newConnection();
        // 創(chuàng)建通道
        Channel channel = connection.createChannel();
        channel.basicPublish("","hello",null,"hello word".getBytes(StandardCharsets.UTF_8));

        // 1.消費(fèi)隊(duì)列名稱 2.開始消息自動(dòng)確認(rèn)機(jī)制 3.消費(fèi)時(shí)回調(diào)接口
        channel.basicConsume("hello", true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                AMQP.BasicProperties properties, byte[] body) throws IOException {
              //  super.handleDelivery(consumerTag, envelope, properties, body);
                System.out.println("消息內(nèi)容:"+new String(body));
            }
        });

        // channel.close();
        // connection.close();
        // con.clone();
    }

image.png