Centos7安裝RabbitMQ與使用（超詳細(xì)），21年大數(shù)據(jù)開發(fā)面經(jīng)分享

這篇具有很好參考價值的文章主要介紹了Centos7安裝RabbitMQ與使用（超詳細(xì)），21年大數(shù)據(jù)開發(fā)面經(jīng)分享。希望對大家有所幫助。如果存在錯誤或未考慮完全的地方，請大家不吝賜教，您也可以點擊"舉報違法"按鈕提交疑問。

public static void main(String[] args) throws Exception{
Connection connection = RabbitMQUtils.getConnection();
Channel channel = connection.createChannel();
/**
* 為了保險起見，防止生產(chǎn)方未啟動隊列未創(chuàng)建的情況下消費方啟動后報404異常，最好在消費方中也聲明創(chuàng)建隊列，注意消費方和生產(chǎn)方聲明的隊列必須完全一致
* 注意：不用考慮隊列是否會重復(fù)創(chuàng)建，在RabbitMQ中如果隊列已經(jīng)存在是不會被重新創(chuàng)建的
*/
channel.queueDeclare(“work-queue”,true,false,false,null);
/**
* 設(shè)置每次抓取的數(shù)據(jù)條數(shù)
* 不設(shè)置：默認(rèn)平分隊列中的消息，如果隊列中中有100條數(shù)據(jù)，那么兩個消費方各抓取50條數(shù)據(jù)進行消費
* 設(shè)置：按照設(shè)置的條數(shù)抓取，如果設(shè)置1，那么消費方從隊列中每次抓取1條數(shù)據(jù)進行消費，消費完成后再抓取1條，直到隊列中沒有消息
*/
channel.basicQos(1);

/**
* 第一個參數(shù):隊列名,所要消費的隊列
* 第二個參數(shù):是否自動確認(rèn)
* true表示自送確認(rèn):消息拿到了就確認(rèn)
* false表示手動確認(rèn):消息處理完成后確認(rèn)
*/

channel.basicConsume(“work-queue”,false,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(new String(body)+envelope.getDeliveryTag());
/**
* 手動確認(rèn)方式
* 第一個參數(shù):包裹的標(biāo)簽（消息的標(biāo)簽，RabbitMQ將每個消息看成是一個包裹），是個整數(shù)
* 第二個參數(shù):是否多條消息批量確認(rèn)，如果第一、二、三…條消息沒有確認(rèn)，后面一條消息確認(rèn)被消費了，那么前面所有的消息都會被確認(rèn)消費了
*/
channel.basicAck(envelope.getDeliveryTag(),false);
}
});
}

}

消費者2：

package com.example.demo.rabbitMQ.work;

import com.example.demo.rabbitMQ.RabbitMQUtils;
import com.rabbitmq.client.*;

import java.io.IOException;

public class ConsumerTwo {
public static void main(String[] args) throws Exception{
Connection connection = RabbitMQUtils.getConnection();
Channel channel = connection.createChannel();
/**
* 為了保險起見，防止生產(chǎn)方未啟動隊列未創(chuàng)建的情況下消費方啟動后報404異常，最好在消費方中也聲明創(chuàng)建隊列，注意消費方和生產(chǎn)方聲明的隊列必須完全一致
* 注意：不用考慮隊列是否會重復(fù)創(chuàng)建，在RabbitMQ中如果隊列已經(jīng)存在是不會被重新創(chuàng)建的
*/
channel.queueDeclare(“work-queue”,true,false,false,null);

/**
* 設(shè)置每次抓取的數(shù)據(jù)條數(shù)
* 不設(shè)置：默認(rèn)平分隊列中的消息，如果隊列中中有100條數(shù)據(jù)，那么兩個消費方各抓取50條數(shù)據(jù)進行消費
* 設(shè)置：按照設(shè)置的條數(shù)抓取，如果設(shè)置1，那么消費方從隊列中每次抓取1條數(shù)據(jù)進行消費，消費完成后再抓取1條，直到隊列中沒有消息
*/
channel.basicQos(1);

channel.basicConsume(“work-queue”,false,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
System.out.println(new String(body)+envelope.getDeliveryTag());
/**
* 手動確認(rèn)方式
* 第一個參數(shù):包裹的標(biāo)簽（消息的標(biāo)簽，RabbitMQ將每個消息看成是一個包裹），是個整數(shù)
* 第二個參數(shù):是否多條消息批量確認(rèn)，如果第一、二、三…條消息沒有確認(rèn)，后面一條消息確認(rèn)被消費了，那么前面所有的消息都會被確認(rèn)消費了
*/
channel.basicAck(envelope.getDeliveryTag(),false);
}
});
}

}

廣播模型-Fanout ============================================================

在廣播模式下，生產(chǎn)者將消息發(fā)送給交換機后，交換機將消息推送給每一個綁定在交換機上的隊列，每一個隊列的消費者都能拿到消息。在Fanout模式中，一條消息，會被所有綁定在該交換機上的隊列消費。

生產(chǎn)者：

package com.example.demo.rabbitMQ.fanout;

import com.example.demo.rabbitMQ.RabbitMQUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Producer {
public static void main(String[] args) throws Exception {
Connection connection = RabbitMQUtils.getConnection();
Channel channel = connection.createChannel();

/**
* 聲明一個Fanout類型的交換機
* 第一個參數(shù):交換機名稱
* 第二個參數(shù):交換機類型:BuiltinExchangeType.FANOUT、BuiltinExchangeType.TOPIC、BuiltinExchangeType.DIRECT
*/
channel.exchangeDeclare(“fanout-exchange”, BuiltinExchangeType.FANOUT);

/**
* 注意：通過channel.queueDeclare()方法創(chuàng)建的Queue綁定在默認(rèn)的交換機上，且BindingKey和隊列名一致
* 第一個參數(shù):隊列名
* 第二個參數(shù):隊列是否持久化，如果為false，rabbitMQ服務(wù)關(guān)閉隊列消失
* 第三個參數(shù):隊列是否為排他隊列，如果為true，隊列僅供創(chuàng)建它的連接使用，當(dāng)前連接關(guān)閉隊列消失
* 第四個參數(shù):隊列是否自動刪除，如果為true，隊列中的消息消費完成，并且消費方關(guān)閉后，隊列自動刪除
*/
channel.queueDeclare(“fanout-queue”,true,false,false,null);

/**
* 將隊列綁定到交換機上
* 第一個參數(shù):隊列名稱
* 第二個參數(shù):交換機名稱
* 第三個參數(shù):BindingKey
*/
channel.queueBind(“fanout-queue”,“fanout-exchange”,“fanout-model”);

/**
* 第一個參數(shù):交換機名稱，如果為“”空串表示使用默認(rèn)交換機，將消息投遞到默認(rèn)交換機，通過默認(rèn)交換機投遞到與之綁定的隊列
* 第二個參數(shù):RoutingKey==BindingKey，在Fanout模式下指定RoutingKey沒有意義，F(xiàn)anout模式下每個與該交換機綁定的隊列都能拿到消息
* 第三個參數(shù):消息的屬性(消息是否持久化、消息存活時間)
* 第四個參數(shù):消息內(nèi)容
*/
channel.basicPublish(“fanout-exchange”,“fanout-model”,null,"Hellow World ".getBytes());
}
}

消費者1：

package com.example.demo.rabbitMQ.fanout;

import com.example.demo.rabbitMQ.RabbitMQUtils;
import com.rabbitmq.client.*;

import java.io.IOException;

public class ConsumerOne {
public static void main(String[] args) throws Exception{
Connection connection = RabbitMQUtils.getConnection();
Channel channel = connection.createChannel();
/**
* 聲明一個Fanout類型的交換機
* 第一個參數(shù):交換機名稱
* 第二個參數(shù):交換機類型
*/
channel.exchangeDeclare(“fanout-exchange”, BuiltinExchangeType.FANOUT);
/**
* 注意：通過channel.queueDeclare()方法創(chuàng)建的Queue綁定在默認(rèn)的交換機上，且BindingKey和隊列名一致
* 第一個參數(shù):隊列名
* 第二個參數(shù):隊列是否持久化，如果為false，連接關(guān)閉隊列消失
* 第三個參數(shù):隊列是否為排他隊列，如果為true，隊列僅供創(chuàng)建它的連接使用，當(dāng)前連接關(guān)閉隊列消失
* 第四個參數(shù):隊列是否自動刪除，如果為true，隊列中的消息消費完成，并且消費方關(guān)閉后，隊列自動刪除
* getQueue()方法可獲取隊列名稱
*/
String queueName = channel.queueDeclare(“fanout-queue-one”, true, false, false, null).getQueue();
/**
* 將隊列綁定到交換機上
* 第一個參數(shù):隊列名稱
* 第二個參數(shù):交換機名稱
* 第三個參數(shù):BindingKey
*/
channel.queueBind(queueName,“fanout-exchange”,“fanout-model-one”);
/**
* 從隊列中消費消息
* 第一個參數(shù):隊列名稱
* 第二個參數(shù):是否自動確認(rèn)
* 第三個參數(shù):消費者
*/
channel.basicConsume(“fanout-queue-one”,false,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(new String(body)+“One”+" "+envelope.getDeliveryTag());
/**
* 確認(rèn)消息是否消費，給隊列反饋
* 第一個參數(shù):包裹（消息）標(biāo)簽
* 第二個參數(shù):是否多條消息批量確認(rèn)，如果第一、二、三…條消息沒有確認(rèn)，后面一條消息確認(rèn)被消費了，那么前面所有的消息都會被確認(rèn)消費了
*/
channel.basicAck(envelope.getDeliveryTag(),true);
}
});
}
}

消費者2：

package com.example.demo.rabbitMQ.fanout;

import com.example.demo.rabbitMQ.RabbitMQUtils;
import com.rabbitmq.client.*;

import java.io.IOException;

public class ConsumerTwo {
public static void main(String[] args) throws Exception{
Connection connection = RabbitMQUtils.getConnection();
Channel channel = connection.createChannel();
/**
* 聲明一個Fanout類型的交換機
* 第一個參數(shù):交換機名稱
* 第二個參數(shù):交換機類型
*/
channel.exchangeDeclare(“fanout-exchange”, BuiltinExchangeType.FANOUT);
/**
* 注意：通過channel.queueDeclare()方法創(chuàng)建的Queue綁定在默認(rèn)的交換機上，且BindingKey和隊列名一致
* 第一個參數(shù):隊列名
* 第二個參數(shù):隊列是否持久化，如果為false，rabbitMQ服務(wù)關(guān)閉隊列消失
* 第三個參數(shù):隊列是否為排他隊列，如果為true，隊列僅供創(chuàng)建它的連接使用，當(dāng)前連接關(guān)閉隊列消失
* 第四個參數(shù):隊列是否自動刪除，如果為true，隊列中的消息消費完成，并且消費方關(guān)閉后，隊列自動刪除
*/
channel.queueDeclare(“fanout-queue-two”,true,false,false,null);
/**
* 將隊列綁定到交換機上
* 第一個參數(shù):隊列名稱
* 第二個參數(shù):交換機名稱
* 第三個參數(shù):BindingKey
*/
channel.queueBind(“fanout-queue-two”,“fanout-exchange”,“fanout-model-two”);
/**
* 從隊列中消費消息
* 第一個參數(shù):隊列名稱
* 第二個參數(shù):是否自動確認(rèn)
* 第三個參數(shù):消費者
*/
channel.basicConsume(“fanout-queue-two”,true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(new String(body)+“Two”);
}
});
}
}

消費者3：

package com.example.demo.rabbitMQ.fanout;

import com.example.demo.rabbitMQ.RabbitMQUtils;
import com.rabbitmq.client.*;

import java.io.IOException;

public class ConsumerThree {
public static void main(String[] args) throws Exception{
Connection connection = RabbitMQUtils.getConnection();
Channel channel = connection.createChannel();
/**
* 聲明一個Fanout類型的交換機
* 第一個參數(shù):交換機名稱
* 第二個參數(shù):交換機類型
*/
channel.exchangeDeclare(“fanout-exchange”, BuiltinExchangeType.FANOUT);
/**
* 注意：通過channel.queueDeclare()方法創(chuàng)建的Queue綁定在默認(rèn)的交換機上，且BindingKey和隊列名一致
* 第一個參數(shù):隊列名
* 第二個參數(shù):隊列是否持久化，如果為false，連接關(guān)閉隊列消失
* 第三個參數(shù):隊列是否為排他隊列，如果為true，隊列僅供創(chuàng)建它的連接使用，當(dāng)前連接關(guān)閉隊列消失
* 第四個參數(shù):隊列是否自動刪除，如果為true，隊列中的消息消費完成，并且消費方關(guān)閉后，隊列自動刪除
* getQueue()方法可獲取隊列名稱
*/
String queueName = channel.queueDeclare(“fanout-queue-one”, true, false, false, null).getQueue();
/**
* 將隊列綁定到交換機上
* 第一個參數(shù):隊列名稱
* 第二個參數(shù):交換機名稱
* 第三個參數(shù):BindingKey
*/
channel.queueBind(queueName,“fanout-exchange”,“fanout-model-one”);
/**
* 設(shè)置每次抓取的數(shù)據(jù)條數(shù)
* 不設(shè)置：默認(rèn)平分隊列中的消息，如果隊列中中有100條數(shù)據(jù)，那么兩個消費方各抓取50條數(shù)據(jù)進行消費
* 設(shè)置：按照設(shè)置的條數(shù)抓取，如果設(shè)置1，那么消費方從隊列中每次抓取1條數(shù)據(jù)進行消費，消費完成后再抓取1條，直到隊列中沒有消息
*/
channel.basicQos(1);
/**
* 從隊列中消費消息
* 第一個參數(shù):隊列名稱
* 第二個參數(shù):是否自動確認(rèn)
* 第三個參數(shù):消費者
*/
channel.basicConsume(queueName,true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(new String(body)+“Three”+" "+envelope.getDeliveryTag());
}
});
}
}

定向模型-Direct ============================================================

在定向模型中，生產(chǎn)者將消息發(fā)送給交換機后，交換機將消息根據(jù)RoutingKey\BindingKey推送到對應(yīng)的隊列上。Exchange不再把消息交給每一個綁定的隊列，而是根據(jù)消息的Routing Key進行判斷，只有隊列的BindingKey與消息的 RoutingKey完全一致，才會接收到消息
生產(chǎn)者：

package com.example.demo.rabbitMQ.direct;

import com.example.demo.rabbitMQ.RabbitMQUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.nio.charset.StandardCharsets;

public class Producer {
public static void main(String[] args) throws Exception{
Connection connection = RabbitMQUtils.getConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(“direct-exchange”, BuiltinExchangeType.DIRECT);
//隊列1
String queue1Name = channel.queueDeclare(“direct-queue-one”, true, false, false, null).getQueue();
channel.queueBind(queue1Name,“direct-exchange”,“direct-model-one”);
//隊列2
String queue2Name = channel.queueDeclare(“direct-queue-two”, true, false, false, null).getQueue();
channel.queueBind(queue2Name,“direct-exchange”,“direct-model-two”);

channel.basicPublish(“direct-exchange”,“direct-model-two”,null,"JAVA是世界上最好的語言 ".getBytes(StandardCharsets.UTF_8));
}
}

消費者1：

package com.example.demo.rabbitMQ.direct;

import com.example.demo.rabbitMQ.RabbitMQUtils;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

public class ConsumerOne {
public static void main(String[] args) throws Exception{
Connection connection = RabbitMQUtils.getConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(“direct-exchange”, BuiltinExchangeType.DIRECT);
String queueName = channel.queueDeclare(“direct-queue-one”, true, false, false, null).getQueue();
channel.queueBind(queueName,“direct-exchange”,“direct-model-one”);

channel.basicConsume(queueName,true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(new String(body, StandardCharsets.UTF_8)+“One”);
}
});
}
}

消費者2：

package com.example.demo.rabbitMQ.direct;

import com.example.demo.rabbitMQ.RabbitMQUtils;
import com.rabbitmq.client.*;

import java.io.IOException;

String queueName = channel.queueDeclare(“direct-queue-two”, true, false, false, null).getQueue();
channel.queueBind(queueName,“direct-exchange”,“direct-model-two”);

主題模型-Topic ============================================================

在主題模式下，生產(chǎn)者將消息發(fā)送給交換機后，交換機根據(jù)RoutingKey\BindingKey匹配隊列，將消息推送到匹配的隊列上。在topic模式下通過#號和*號進行模糊匹配,通過.進行分割,#號表示可能有一個或多個單詞，也可能沒有；*號表示有且僅有一個單詞

生產(chǎn)者：

package com.example.demo.rabbitMQ.topic;

import com.example.demo.rabbitMQ.RabbitMQUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Producer {
public static void main(String[] args) throws Exception{
Connection connection = RabbitMQUtils.getConnection();
Channel channel = connection.createChannel();
/**
* 聲明一個TOPIC類型的交換機
* 第一個參數(shù):交換機名稱
* 第二個參數(shù):交換機類型:BuiltinExchangeType.FANOUT、BuiltinExchangeType.TOPIC、BuiltinExchangeType.DIRECT
*/
channel.exchangeDeclare(“topic-exchange”, BuiltinExchangeType.TOPIC);
/**
* 聲明隊列1:
* 注意：通過channel.queueDeclare()方法創(chuàng)建的Queue綁定在默認(rèn)的交換機上，且BindingKey和隊列名一致
* 第一個參數(shù):隊列名
* 第二個參數(shù):隊列是否持久化，如果為false，rabbitMQ服務(wù)關(guān)閉隊列消失
* 第三個參數(shù):隊列是否為排他隊列，如果為true，隊列僅供創(chuàng)建它的連接使用，當(dāng)前連接關(guān)閉隊列消失
* 第四個參數(shù):隊列是否自動刪除，如果為true，隊列中的消息消費完成，并且消費方關(guān)閉后，隊列自動刪除
*/
String queue1Name = channel.queueDeclare(“topic-queue-one”, true, false, false, null).getQueue();
/**
* 將隊列綁定到交換機上
* 第一個參數(shù):隊列名稱
* 第二個參數(shù):交換機名稱
* 第三個參數(shù):BindingKey,在topic模式下通過#和*進行模糊匹配,通過.進行分割,#號表示可能有一個或多個單詞，也可能沒有；*號表示有且僅有一個單詞
*/
channel.queueBind(queue1Name,“topic-exchange”,“topic-model.#”);

/**
* 聲明隊列2:
* 注意：通過channel.queueDeclare()方法創(chuàng)建的Queue綁定在默認(rèn)的交換機上，且BindingKey和隊列名一致
* 第一個參數(shù):隊列名
* 第二個參數(shù):隊列是否持久化，如果為false，rabbitMQ服務(wù)關(guān)閉隊列消失
* 第三個參數(shù):隊列是否為排他隊列，如果為true，隊列僅供創(chuàng)建它的連接使用，當(dāng)前連接關(guān)閉隊列消失
* 第四個參數(shù):隊列是否自動刪除，如果為true，隊列中的消息消費完成，并且消費方關(guān)閉后，隊列自動刪除
*/
String queue2Name = channel.queueDeclare(“topic-queue-two”, true, false, false, null).getQueue();
/**
* 將隊列綁定到交換機上
* 第一個參數(shù):隊列名稱
* 第二個參數(shù):交換機名稱
* 第三個參數(shù):BindingKey,在topic模式下通過#和*進行模糊匹配,通過.進行分割,#號表示可能有一個或多個單詞，也可能沒有；*號表示有且僅有一個單詞
*/
channel.queueBind(queue2Name,“topic-exchange”,“topic-model.*”);

/**
* 聲明隊列3:
* 注意：通過channel.queueDeclare()方法創(chuàng)建的Queue綁定在默認(rèn)的交換機上，且BindingKey和隊列名一致
* 第一個參數(shù):隊列名
* 第二個參數(shù):隊列是否持久化，如果為false，rabbitMQ服務(wù)關(guān)閉隊列消失
* 第三個參數(shù):隊列是否為排他隊列，如果為true，隊列僅供創(chuàng)建它的連接使用，當(dāng)前連接關(guān)閉隊列消失
* 第四個參數(shù):隊列是否自動刪除，如果為true，隊列中的消息消費完成，并且消費方關(guān)閉后，隊列自動刪除
*/
String queue3Name = channel.queueDeclare(“topic-queue-three”, true, false, false, null).getQueue();
/**
* 將隊列綁定到交換機上
* 第一個參數(shù):隊列名稱
* 第二個參數(shù):交換機名稱
* 第三個參數(shù):BindingKey,在topic模式下通過#和*進行模糊匹配,通過.進行分割,#號表示可能有一個或多個單詞，也可能沒有；*號表示有且僅有一個單詞
*/
channel.queueBind(queue3Name,“topic-exchange”,“topic-model.three.#”);

/**
* 第一個參數(shù):交換機名稱，如果為“”空串表示使用默認(rèn)交換機，將消息投遞到默認(rèn)交換機，通過默認(rèn)交換機投遞到與之綁定的隊列
* 第二個參數(shù):RoutingKey==BindingKey，在topic模式下通過#和*進行模糊匹配,通過.進行分割,#號表示可能有一個或多個單詞，也可能沒有；*號表示有且僅有一個單詞
* 第三個參數(shù):消息的屬性(消息是否持久化、消息存活時間)
* 第四個參數(shù):消息內(nèi)容
*/
channel.basicPublish(“topic-exchange”,“topic-model.three.java”,null,“java是世界上最優(yōu)秀的語言”.getBytes());

}
}

消費者1：

package com.example.demo.rabbitMQ.topic;

import com.example.demo.rabbitMQ.RabbitMQUtils;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

public class ConsumerOne {
public static void main(String[] args) throws Exception {
Connection connection = RabbitMQUtils.getConnection();
Channel channel = connection.createChannel();
/**
* 為了保險起見，防止生產(chǎn)方未啟動隊列\(zhòng)交換機未創(chuàng)建消的情況下費方啟動后報404異常，最好在消費方中也聲明創(chuàng)建隊列，注意消費方和生產(chǎn)方聲明的隊列\(zhòng)交換機必須完全一致
* 注意：不用考慮隊列\(zhòng)交換機是否會重復(fù)創(chuàng)建，在RabbitMQ中如果隊列已經(jīng)存在是不會被重新創(chuàng)建的
*/

//聲明交換機
channel.exchangeDeclare(“topic-exchange”, BuiltinExchangeType.TOPIC);
//聲明隊列
String queue1Name = channel.queueDeclare(“topic-queue-one”, true, false, false, null).getQueue();
//將隊列綁定到交換機上
channel.queueBind(queue1Name,“topic-exchange”,“topic-model.#”);
//設(shè)置每次抓取的數(shù)據(jù)條數(shù)
channel.basicQos(1);
/**
* 第一個參數(shù):被消費的隊列名
* 第二個參數(shù):是否自動確認(rèn)
* 第三個參數(shù):使用默認(rèn)的消費者
*/
channel.basicConsume(queue1Name,true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(new String(body, StandardCharsets.UTF_8));
//在手動確認(rèn)的情況下使用
/**
* 確認(rèn)消息是否消費，給隊列反饋
* 第一個參數(shù):包裹（消息）標(biāo)簽
* 第二個參數(shù):是否多條消息批量確認(rèn)，如果第一、二、三…條消息沒有確認(rèn)，后面一條消息確認(rèn)被消費了，那么前面所有的消息都會被確認(rèn)消費了
*/
//channel.basicAck(envelope.getDeliveryTag(),false);
}
});
}
}

消費者2：

package com.example.demo.rabbitMQ.topic;

import com.example.demo.rabbitMQ.RabbitMQUtils;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

public class ConsumerTwo {
public static void main(String[] args) throws Exception {
Connection connection = RabbitMQUtils.getConnection();
Channel channel = connection.createChannel();
/**
* 為了保險起見，防止生產(chǎn)方未啟動隊列\(zhòng)交換機未創(chuàng)建消的情況下費方啟動后報404異常，最好在消費方中也聲明創(chuàng)建隊列，注意消費方和生產(chǎn)方聲明的隊列\(zhòng)交換機必須完全一致
* 注意：不用考慮隊列\(zhòng)交換機是否會重復(fù)創(chuàng)建，在RabbitMQ中如果隊列已經(jīng)存在是不會被重新創(chuàng)建的
*/

//聲明交換機
channel.exchangeDeclare(“topic-exchange”, BuiltinExchangeType.TOPIC);
//聲明隊列
String queue2Name = channel.queueDeclare(“topic-queue-two”, true, false, false, null).getQueue();
//將隊列綁定到交換機上
channel.queueBind(queue2Name,“topic-exchange”,“topic-model.*”);
//設(shè)置每次抓取的數(shù)據(jù)條數(shù)
channel.basicQos(1);
/**
* 第一個參數(shù):被消費的隊列名
* 第二個參數(shù):是否自動確認(rèn)
* 第三個參數(shù):使用默認(rèn)的消費者
*/
channel.basicConsume(queue2Name,false,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(new String(body, StandardCharsets.UTF_8));
//在手動確認(rèn)的情況下使用
/**
* 確認(rèn)消息是否消費，給隊列反饋
* 第一個參數(shù):包裹（消息）標(biāo)簽
* 第二個參數(shù):是否多條消息批量確認(rèn)，如果第一、二、三…條消息沒有確認(rèn)，后面一條消息確認(rèn)被消費了，那么前面所有的消息都會被確認(rèn)消費了
*/
channel.basicAck(envelope.getDeliveryTag(),false);
}
});
}
}

消費者3：

package com.example.demo.rabbitMQ.topic;

import com.example.demo.rabbitMQ.RabbitMQUtils;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

public class ConsumerThree {
public static void main(String[] args) throws Exception {
Connection connection = RabbitMQUtils.getConnection();
Channel channel = connection.createChannel();
/**
* 為了保險起見，防止生產(chǎn)方未啟動隊列\(zhòng)交換機未創(chuàng)建消的情況下費方啟動后報404異常，最好在消費方中也聲明創(chuàng)建隊列，注意消費方和生產(chǎn)方聲明的隊列\(zhòng)交換機必須完全一致
* 注意：不用考慮隊列\(zhòng)交換機是否會重復(fù)創(chuàng)建，在RabbitMQ中如果隊列已經(jīng)存在是不會被重新創(chuàng)建的
*/

//聲明交換機
channel.exchangeDeclare(“topic-exchange”, BuiltinExchangeType.TOPIC);
//聲明隊列
String queue3Name = channel.queueDeclare(“topic-queue-two”, true, false, false, null).getQueue();
//將隊列綁定到交換機上
channel.queueBind(queue3Name,“topic-exchange”,“topic-model”);
//設(shè)置每次抓取的數(shù)據(jù)條數(shù)
channel.basicQos(1);
/**
* 第一個參數(shù):被消費的隊列名
* 第二個參數(shù):是否自動確認(rèn)
* 第三個參數(shù):使用默認(rèn)的消費者
*/
channel.basicConsume(queue3Name,false,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(new String(body, StandardCharsets.UTF_8));
//在手動確認(rèn)的情況下使用
/**
* 確認(rèn)消息是否消費，給隊列反饋
* 第一個參數(shù):包裹（消息）標(biāo)簽
* 第二個參數(shù):是否多條消息批量確認(rèn)，如果第一、二、三…條消息沒有確認(rèn)，后面一條消息確認(rèn)被消費了，那么前面所有的消息都會被確認(rèn)消費了
*/
channel.basicAck(envelope.getDeliveryTag(),false);
}
});
}
}

SpringBoot+RabbitMQ

第一步：先導(dǎo)依賴

org.springframework.boot spring-boot-starter-amqp com.alibaba fastjson 1.2.4

第二步：配置RabbitMQ

spring:
rabbitmq:
host: 192.168.119.134
port: 5672
username: rabbit
password: 123456

第三步：配置RabbitMQ使用的序列化

// 消息的消費方j(luò)son數(shù)據(jù)的反序列化
@Bean
public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(
ConnectionFactory connectionFactory){
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setMessageConverter(new Jackson2JsonMessageConverter());
return factory;
}

// 定義使用json的方式轉(zhuǎn)換數(shù)據(jù)
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate amqpTemplate = new RabbitTemplate();
amqpTemplate.setConnectionFactory(connectionFactory);
amqpTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
return amqpTemplate;
}

第四步：根據(jù)情況選擇合適的模型

一、簡單模型 ============================================================

特點：隊列是綁定在默認(rèn)交換機上，BindingKey就是隊列名

消息投遞方：
在SpringBoot中即可使用RabbitTemplate進行消息的投遞，也可使用原生的方式進行消息投遞

package com.example.demo.rabbitMQ.springBootrabbitMQ.simple;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
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;

@RestController
@RequestMapping(“/rabbit”)
public class Producer {
@Autowired
private RabbitTemplate rabbitTemplate;

@GetMapping(“simple”)
public String sendMessage(){
rabbitTemplate.convertAndSend(“simple-queue”,“落霞與孤鶩齊飛，秋水共長天一色”);
return “成功”;
}
}

消息消費方
在SpringBoot中即可使用注解@RabbitListener進行消息的投遞，也可使用原生的方式進行消息投遞

package com.example.demo.rabbitMQ.springBootrabbitMQ.simple;

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

@Component
public class Consumer {
/**
* 1、@RabbitListener(queues = {“topic-queue-one”} ),一個@RabbitListener就是一個消費者
* 2、使用queues，如果隊列不存在就會報異常，使用queuesToDeclare，如果隊列不存在就創(chuàng)建隊列
* 3、注意：SpringBoot提供了一個很好的消息確認(rèn)機制，如果消費方消費的過程中有異常，一定要像service層一樣拋出異常不能捕捉
* 4、監(jiān)聽隊列，如果有消息就進行消費
* 5、簡單模型和工作模型的queue都是綁定在默認(rèn)交換機上
*/
@RabbitListener(queuesToDeclare = {@Queue(name = “simple-queue”,durable = “true”)})
public void getMessage(String message){
System.out.println("springBoot+rabbitMQ: "+message);
}
}

二、Work模型 ============================================================

特點：工作模型只是在簡單模型的基礎(chǔ)上增加了多個消費者

消息投遞方

package com.example.demo.rabbitMQ.springBootrabbitMQ.work;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping(“/rabbit”)
public class Producer {
@Autowired
private RabbitTemplate rabbitTemplate;

@RequestMapping(value = “/work”,method = RequestMethod.GET)
public void sendMessage(){
rabbitTemplate.convertAndSend(“work-queue-one”,“祖國您好！”);
}
}

消息消費方

package com.example.demo.rabbitMQ.springBootrabbitMQ.work;

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

@Component
public class Consumer {

/**
* 1、@RabbitListener(queues = {“topic-queue-one”} ),一個@RabbitListener就是一個消費者
* 2、使用queues，如果隊列不存在就會報異常，使用queuesToDeclare，如果隊列不存在就創(chuàng)建隊列
* 3、注意：SpringBoot提供了一個很好的消息確認(rèn)機制，如果消費方消費的過程中有異常，一定要像service層一樣拋出異常不能捕捉
* 4、監(jiān)聽隊列，如果有消息就進行消費
* 5、簡單模型和工作模型的queue都是綁定在默認(rèn)交換機上,工作模型就是在簡單模型的基礎(chǔ)上增加了多個消費者
*/

@RabbitListener(queuesToDeclare = @Queue(“work-queue-one”))
public void getMessageOne(String message){
System.out.println(message);
}

@RabbitListener(queuesToDeclare = @Queue(“work-queue-two”))
public void getMessageTwo(String message){
System.out.println(message);
}
}

三、Fanout模型 ============================================================

特點：凡是綁定在該交換機上的隊列都會收到消息，無關(guān)路由鍵

消息投遞方

package com.example.demo.rabbitMQ.springBootrabbitMQ.fanout;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping(“rabbit”)
public class Producer {
@Autowired
private RabbitTemplate rabbitTemplate;

@RequestMapping(“fanout”)
//fanout類型的交換機無關(guān)路由鍵
public void sendMessage(){
rabbitTemplate.convertAndSend(“fanout-exchange”,“”,“這是fanout類型的交換機”);
}
}

消息消費方

package com.example.demo.rabbitMQ.springBootrabbitMQ.fanout;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;

自我介紹一下，小編13年上海交大畢業(yè)，曾經(jīng)在小公司待過，也去過華為、OPPO等大廠，18年進入阿里一直到現(xiàn)在。

深知大多數(shù)大數(shù)據(jù)工程師，想要提升技能，往往是自己摸索成長或者是報班學(xué)習(xí)，但對于培訓(xùn)機構(gòu)動則幾千的學(xué)費，著實壓力不小。自己不成體系的自學(xué)效果低效又漫長，而且極易碰到天花板技術(shù)停滯不前！

因此收集整理了一份《2024年大數(shù)據(jù)全套學(xué)習(xí)資料》，初衷也很簡單，就是希望能夠幫助到想自學(xué)提升又不知道該從何學(xué)起的朋友。
Centos7安裝RabbitMQ與使用（超詳細(xì)），21年大數(shù)據(jù)開發(fā)面經(jīng)分享,2024年程序員學(xué)習(xí),rabbitmq,大數(shù)據(jù),ruby

既有適合小白學(xué)習(xí)的零基礎(chǔ)資料，也有適合3年以上經(jīng)驗的小伙伴深入學(xué)習(xí)提升的進階課程，基本涵蓋了95%以上大數(shù)據(jù)開發(fā)知識點，真正體系化！

由于文件比較大，這里只是將部分目錄大綱截圖出來，每個節(jié)點里面都包含大廠面經(jīng)、學(xué)習(xí)筆記、源碼講義、實戰(zhàn)項目、講解視頻，并且后續(xù)會持續(xù)更新

如果你覺得這些內(nèi)容對你有幫助，可以添加VX：vip204888 （備注大數(shù)據(jù)獲?。?/strong>

te.convertAndSend(“fanout-exchange”,“”,“這是fanout類型的交換機”);
}
}

消息消費方

package com.example.demo.rabbitMQ.springBootrabbitMQ.fanout;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;

自我介紹一下，小編13年上海交大畢業(yè)，曾經(jīng)在小公司待過，也去過華為、OPPO等大廠，18年進入阿里一直到現(xiàn)在。

深知大多數(shù)大數(shù)據(jù)工程師，想要提升技能，往往是自己摸索成長或者是報班學(xué)習(xí)，但對于培訓(xùn)機構(gòu)動則幾千的學(xué)費，著實壓力不小。自己不成體系的自學(xué)效果低效又漫長，而且極易碰到天花板技術(shù)停滯不前！

因此收集整理了一份《2024年大數(shù)據(jù)全套學(xué)習(xí)資料》，初衷也很簡單，就是希望能夠幫助到想自學(xué)提升又不知道該從何學(xué)起的朋友。
[外鏈圖片轉(zhuǎn)存中…(img-kdSdEgbK-1712513731160)]
[外鏈圖片轉(zhuǎn)存中…(img-A5rWcrQG-1712513731161)]
[外鏈圖片轉(zhuǎn)存中…(img-bD5ZrjXL-1712513731161)]
[外鏈圖片轉(zhuǎn)存中…(img-BhvSH84G-1712513731162)]
[外鏈圖片轉(zhuǎn)存中…(img-hXo1vlAm-1712513731162)]

既有適合小白學(xué)習(xí)的零基礎(chǔ)資料，也有適合3年以上經(jīng)驗的小伙伴深入學(xué)習(xí)提升的進階課程，基本涵蓋了95%以上大數(shù)據(jù)開發(fā)知識點，真正體系化！

由于文件比較大，這里只是將部分目錄大綱截圖出來，每個節(jié)點里面都包含大廠面經(jīng)、學(xué)習(xí)筆記、源碼講義、實戰(zhàn)項目、講解視頻，并且后續(xù)會持續(xù)更新

如果你覺得這些內(nèi)容對你有幫助，可以添加VX：vip204888 （備注大數(shù)據(jù)獲取）
[外鏈圖片轉(zhuǎn)存中…(img-iMitFADB-1712513731162)]文章來源地址http://www.zghlxwxcb.cn/news/detail-853141.html
到了這里，關(guān)于Centos7安裝RabbitMQ與使用（超詳細(xì)），21年大數(shù)據(jù)開發(fā)面經(jīng)分享的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！