探索解析微服務下的RabbitMQ
概覽
本文主要介紹如何使用RabbitMQ訊息代理來實現分散式系統之間的通訊，從而促進微服務的鬆耦合。
RabbitMQ ，也被稱為開源訊息代理，它支援多種訊息協議，並且可以部署在分散式系統上。它輕量級，便於部署應用程式。它主要充當一個佇列，其中輸入的訊息可以首先被操作。RabbitMQ可以在許多作業系統和雲環境中執行，併為大多數流行語言提供了廣泛的開發工具。它是生產者-消費者模式，生產者發出資訊，消費者消費資訊。RabbitMQ的主要特點如下：
非同步訊息
分散式部署
管理和監控
企業和雲端計算
安裝
對於RabbitMQ，首先需要在系統中安裝ErLang，因為RabbitMQ是用ErLang語言編寫的。安裝Erlang之後，你可以通過下面的介紹從它的官網下載最新版本的 RabbitMQ 。
在微服務中使用RabbitMQ
在您的微服務體系結構中，RabbitMQ是實現訊息佇列的最簡單的免費的可用選項之一。這些佇列模式有助於解耦各個微服務之間的通訊來增加應用程式的彈性。我們可以將這些佇列用於各種目的，比如核心微服務之間的互動、微服務的解耦、實現故障轉移機制，以及通過訊息代理髮送電子郵件通知。
無論在哪裡，只要有兩個或兩個以上的核心模組需要相互通訊，我們就不應該進行直接的HTTP呼叫，因為它們會使核心層產生緊耦合，並且當每個核心模組有更多例項時將很難管理。而且每當服務宕機時，HTTP呼叫模式就會失敗，因為在服務重啟之後，我們將無法跟蹤舊的HTTP請求呼叫。這就產生了對RabbitMQ的需求。
在微服務中設定RabbitMQ
在微服務架構中，為了演示，我們將使用一個可以通過任何核心微服務傳送電子郵件通知的示例模式。在這種模式下，我們將有一個可以存在任何核心微服務的生產者，它將生成電子郵件內容並將其傳送到佇列。然後，這個電子郵件內容由總是在等待佇列中新訊息的消費者來處理。
請注意，由於正在使用Spring Boot構建微服務，因此我們將為Spring提供配置。
1）生產者： 這一層負責生成電子郵件內容，並將此內容傳送給RabbitMQ中的訊息代理。
a）在properties檔案中，我們需要配置佇列名和交換型別，以及安裝RabbitMQ伺服器的主機和埠。
queue.name=messagequeue
fanout.exchange=messagequeue-exchange
spring.rabbitmq.host: localhost
spring.rabbitmq.port: 5672
spring.rabbitmq.username: guest
spring.rabbitmq.password: guest
b）我們需要建立一個配置類，它將使用佇列名和交換型別將佇列繫結到微服務模組。
@Configuration
public class RabbitConfiguration {
@Value("${fanout.exchange}")
private String fanoutExchange;
@Value("${queue.name}")
private String queueName;
@Bean
Queue queue() {
return new Queue(queueName, true);
}
@Bean
FanoutExchange exchange() {
return new FanoutExchange(fanoutExchange);
}
@Bean
Binding binding(Queue queue, FanoutExchange exchange) {
return BindingBuilder.bind(queue).to(exchange);
}
}
c）最後，我們需要一個工具類，它將使用Spring框架提供的RabbitTemplate將實際的電子郵件內容傳送到佇列中。
@Component
public class QueueProducer {
protected Logger logger = LoggerFactory.getLogger(getClass());
@Value("${fanout.exchange}")
private String fanoutExchange;
private final RabbitTemplate rabbitTemplate;
@Autowired
public QueueProducer(RabbitTemplate rabbitTemplate) {
super();
this.rabbitTemplate = rabbitTemplate;
}
public void produce(NotificationRequestDTO notificationDTO) throws Exception {
logger.info("Storing notification...");
rabbitTemplate.setExchange(fanoutExchange);
rabbitTemplate.convertAndSend(new ObjectMapper().writeValueAsString(notificationDTO));
logger.info("Notification stored in queue sucessfully");
}
}
d）然後，您可以在模組的任何地方呼叫這個produce方法。
{
queueProducer.produce(notificationDTO);
}
2） 消費者: 這一層負責使用FIFO方法從RabbitMQ訊息代理中消費訊息，然後執行與電子郵件相關的操作。
a）在這個properties檔案中，我們需要配置佇列名和交換型別，以及安裝RabbitMQ伺服器的主機和埠。
queue.name=messagequeue
fanout.exchange=messagequeue-exchange
spring.rabbitmq.host: localhost
spring.rabbitmq.port: 5672
spring.rabbitmq.username: guest
spring.rabbitmq.password: guest
b）我們需要建立一個配置類，它將使用佇列名和交換型別將佇列繫結到微服務模組。此外，在消費者的RabbitMQ配置中，我們需要建立一個充當消費者的MessageListenerAdapter bean，它始終偵聽從佇列中傳入的訊息。這個MessageListenerAdapter將有一個帶有消費者工具類和defaultListenerMethod的有參建構函式，在這裡我們可以指定與電子郵件相關的操作。
@Configuration
public class RabbitConfiguration {
private static final String LISTENER_METHOD = "receiveMessage";
@Value("${queue.name}")
private String queueName;
@Value("${fanout.exchange}")
private String fanoutExchange;
@Bean
Queue queue() {
return new Queue(queueName, true);
}
@Bean
FanoutExchange exchange() {
return new FanoutExchange(fanoutExchange);
}
@Bean
Binding binding(Queue queue, FanoutExchange exchange) {
return BindingBuilder.bind(queue).to(exchange);
}
@Bean
SimpleMessageListenerContainer container(ConnectionFactory connectionFactory,
MessageListenerAdapter listenerAdapter) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setQueueNames(queueName);
container.setMessageListener(listenerAdapter);
return container;
}
@Bean
MessageListenerAdapter listenerAdapter(QueueConsumer consumer) {
return new MessageListenerAdapter(consumer, LISTENER_METHOD);
}
}
c）然後，需要建立具有特定訊息偵聽器方法的 QueueConsumer類，在該類中我們可以進行實際傳送電子郵件的操作。
@Component
public class QueueConsumer {
@Autowired
MailServiceImpl mailServiceImpl;
protected Logger logger = LoggerFactory.getLogger(getClass());
public void receiveMessage(String message) {
logger.info("Received (String) " + message);
processMessage(message);
}
public void receiveMessage(byte[] message) {
String strMessage = new String(message);
logger.info("Received (No String) " + strMessage);
processMessage(strMessage);
}
private void processMessage(String message) {
try {
MailDTO mailDTO = new ObjectMapper().readValue(message, MailDTO.class);
ValidationUtil.validateMailDTO(mailDTO);
mailServiceImpl.sendMail(mailDTO, null);
} catch (JsonParseException e) {
logger.warn("Bad JSON in message: " + message);
} catch (JsonMappingException e) {
logger.warn("cannot map JSON to NotificationRequest: " + message);
} catch (Exception e) {
logger.error(e.getMessage());
}
}
}
總結
通過使用RabbitMQ，您可以避免服務之間直接的HTTP呼叫，並消除核心微服務的緊密耦合。這將幫助您在更高級別上實現微服務的可伸縮性，並在微服務之間新增故障轉移機制。
