java RabbitMQ+SpringAMPQ:在SimpleMessageListenerContainer中接收多个队列
我目前正在学习RabbitMQ+SpringAMQ的一些知识,我试图让我的SimpleMessageListenerContainer能够读取发送到两个队列的消息,但只有一个队列正在接收。如果你对代码有任何进一步的建议,或者有任何更好的方法,请告诉我
编辑: 问题是 代码没有将消息发送到两个队列,而是只发送到一个队列
基本上,我从两个队列接收数据,一个在数据库中插入数据,另一个将接收验证请求。我希望稍后向路由密钥和交换发送响应
这是我的申请表。爪哇:
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.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
@SpringBootApplication
public class ChallengeApplication {
public static void main(String[] args) {
SpringApplication.run(ChallengeApplication.class, args);
}
static final String responseExchange = "response.exchange";
static final String routingKey = "response.routing.key";
static final String queueInsertion = "insertion.queue";
static final String queueValidation = "validation.queue";
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost("127.0.0.1");
connectionFactory.setVirtualHost("/");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
return connectionFactory;
}
@Bean
Queue queueInsertion() {
return new Queue(queueInsertion, true);
}
@Bean
Queue queueValidation() {
return new Queue(queueValidation, true);
}
@Bean
DirectExchange exchange() {
return new DirectExchange(responseExchange);
}
@Bean
Binding bindingInsertion(@Qualifier("queueInsertion") Queue queue, DirectExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(queueInsertion);
}
@Bean
Binding queueValidation(@Qualifier("queueValidation") Queue queue, DirectExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(queueValidation);
}
@Bean
SimpleMessageListenerContainer containerValidation(ConnectionFactory connectionFactory,
MessageListenerAdapter listenerAdapter) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setConcurrentConsumers(3);
container.setQueueNames(queueValidation, queueInsertion);
container.setMessageListener(listenerAdapter);
return container;
}
@Bean
MessageListenerAdapter listenerAdapter(Receiver receiver) {
return new MessageListenerAdapter(receiver, "receiveMessage");
}
}
正如您所看到的,我的SimpleMessageListenerContainer在setQueueNames中包含了我想要收听的两个队列。我的MessageListenerAdapter将其发送给我的接收者。对于receiveMessage方法,另一个问题是如何读取接收消息的队列?这是可能的,或者我应该使用@RabbitListener来指向每个地方
接受者。爪哇
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.regex.Pattern;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.stereotype.Component;
import com.axur.challenge.DAO.WhitelistDAO;
import com.axur.challenge.formatters.InputData;
import com.axur.challenge.model.Whitelist;
import com.google.gson.Gson;
import com.google.gson.JsonParseException;
@Component
public class Receiver {
@Autowired
private WhitelistDAO whitelistDAO;
private CountDownLatch latch = new CountDownLatch(1);
public void receiveMessage(String message) {
try {
InputData inputData = new Gson().fromJson(message, InputData.class);
String url = inputData.getUrl();
if (url == null) {
receivedInsertion(inputData);
} else {
receivedValidation(inputData);
}
} catch (JsonParseException e) {
System.out.println("It was not possible to read the input");
}
latch.countDown();
}
public void receivedInsertion(InputData inputData) {
Whitelist whitelist = new Whitelist();
String client = inputData.getClient();
if (client == null) {
whitelist.setClient("global");
} else {
whitelist.setClient(inputData.getClient());
}
whitelist.setRegex(inputData.getRegex());
System.out.println("Whitelist: " + whitelist);
try {
Whitelist selectedWhitelist = whitelistDAO.getSpecificWhitelist(whitelist.getClient(), whitelist.getRegex());
System.out.println("selectedWhitelist: " + selectedWhitelist);
if (selectedWhitelist == null || !selectedWhitelist.equals(whitelist)) {
whitelistDAO.insertWhitelist(whitelist);
System.out.println("Whitelist Added");
} else {
System.out.println("Already added before");
}
} catch (DataAccessException dae) {
System.out.println("Whitelist NOT Added");
System.err.println(dae);
}
}
public void receivedValidation(InputData inputData) {
//getting all regex from this client
List<Whitelist> listWhitelist = whitelistDAO.getWhitelist(inputData.getClient(), inputData.getRegex());
boolean match = false;
//verify if the regex works for the url provided
if (listWhitelist != null) {
int index = 0;
while (match == false && index < listWhitelist.size()) {
match = checkRegex(listWhitelist.get(index).getRegex(), inputData.getUrl());
index++;
}
}
if (match == true) {
} else {
}
}
public boolean checkRegex(String regex, String url) {
return Pattern.compile(regex).matcher(url).find();
}
public CountDownLatch getLatch() {
return latch;
}
}
这是跑步者。java,我将消息发送到每个队列以进行测试:
import java.util.concurrent.TimeUnit;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
@Component
public class Runner implements CommandLineRunner {
private final RabbitTemplate rabbitTemplate;
private final Receiver receiver;
public Runner(Receiver receiver, RabbitTemplate rabbitTemplate) {
this.receiver = receiver;
this.rabbitTemplate = rabbitTemplate;
}
@Override
public void run(String... args) throws Exception {
System.out.println("Sending message...");
rabbitTemplate.convertAndSend(ChallengeApplication.responseExchange, ChallengeApplication.queueInsertion, "{'client':null, 'regex':'[a-z]'}");
rabbitTemplate.convertAndSend(ChallengeApplication.responseExchange, ChallengeApplication.queueValidation, "{'client':null, 'regex':'[a-z]'}");
//rabbitTemplate.convertAndSend(ChallengeApplication.responseExchange, ChallengeApplication.routingKey, "{'client':null, 'regex':'[a-z]'}");
receiver.getLatch().await(3, TimeUnit.SECONDS);
}
}
谢谢你抽出时间
# 1 楼答案
在您的情况下,我认为您应该使用RabbitListner并将队列的名称作为参数传递,如下所示:
Ans您应该将rabbitTemplate作为bean注入配置类,在这种情况下,您可以在队列存在时使用许多listner:
# 2 楼答案
如果不能或不想使用注释,则需要为接收器实现
MessageListener
接口。然后,您将收到一个Message
对象,该对象包含额外的元数据下面是一个例子:
您可以通过以下方式注册接收器: