除錯了一天了，無法解決rabbitmq的廣播模式，兩個消費者，一條A消費者接受到，一條B消費者接收到，不能同時接收到，必須是同一個queue,求大神解答，代碼如下：
rabbitTemplate 和 cachingConnectionFactory工廠連接類
@Configuration
@EnableRabbit
public class RabbitConfiguration {
@Autowired
RabbitProperties properties;
@Bean
public CachingConnectionFactory cachingConnectionFactory(){
CachingConnectionFactory factory = new CachingConnectionFactory();
factory.setAddresses(properties.getAddresses());
factory.setUsername(properties.getUsername());
factory.setPassword(properties.getPassword());
factory.setVirtualHost(properties.getVirtualHost());
factory.setPublisherConfirms(true);
return factory;
}
@Bean
public RabbitAdmin rabbitAdmin() {
return new RabbitAdmin(cachingConnectionFactory());
}
@Bean
public RabbitTemplate rabbitTemplate(CachingConnectionFactory cachingConnectionFactory){
RabbitTemplate template = new RabbitTemplate(cachingConnectionFactory);
template.setMessageConverter(new Jackson2JsonMessageConverter());
return template;
}
}
queue exchange listener 宣告類
@Configuration("syncRabbitConfiguration")
@EnableRabbit
public class RabbitConfiguration {
public static final String QUEUE = "external_socket.request";
public static final String EXCHANGE = "external_socket.request";
@Autowired
CachingConnectionFactory cachingConnectionFactory;
@Autowired
RabbitAdmin rabbitAdmin;
@Autowired
SocketConsumer consumer;
@Bean
public FanoutExchange socketExchange(){
return new FanoutExchange(EXCHANGE);
}
@Bean
public Queue socketQueue(){
return new Queue(QUEUE);
}
@Bean
public Binding socketBinding(Queue socketQueue,FanoutExchange socketExchange){
return BindingBuilder.bind(socketQueue).to(socketExchange);
}
@Bean
public SimpleMessageListenerContainer smsContainer(Queue socketQueue){
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(cachingConnectionFactory);
container.setAcknowledgeMode(AcknowledgeMode.NONE);
container.setExposeListenerChannel(true);
container.setMaxConcurrentConsumers(1);
container.setConcurrentConsumers(1);
container.setRabbitAdmin(rabbitAdmin);
container.setQueues(socketQueue);
container.setMessageListener(consumer);
container.setRecoveryInterval(MQConstant.Rabbit.DEFAULT_RECOVERY_INTERVAL);
return container;
}
}
其中container.setAcknowledgeMode(AcknowledgeMode.NONE);
container.setExposeListenerChannel(true);
container.setMaxConcurrentConsumers(1);
container.setConcurrentConsumers(1);
這幾個引數我都調整過，沒用，下面是消費者代碼
@Service
public class SocketConsumer implements ChannelAwareMessageListener {
@Override
public void onMessage(Message message, Channel channel) throws Exception {
String body = null;
try {
body = new String(message.getBody(),"UTF-8");
SimpleDataDto dto = JSONObject.parseObject(body, SimpleDataDto.class);
System.out.println(dto.getSyncType()+":收到訊息");
if(dto.getSyncType() == null){
List<SocketIOClient> allClient = SessionManager.getAllClient();
for (SocketIOClient socketIOClient : allClient) {
socketIOClient.sendEvent(dto.getDataType(),dto.getData());
}
}else{
SocketIOClient client = SessionManager.getClient(dto.getSyncType());
if(client !=null)
client.sendEvent(dto.getDataType(),dto.getData());
}
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}catch (Exception e){
e.printStackTrace();
}
}
}
發送代碼就一行
rabbitTemplate.convertAndSend(RabbitConfiguration.EXCHANGE,"",dataDto);
通過自帶的web管理段發布訊息也是一樣的效果
下面是管理端的截圖
exchange
queue
實在找不到辦法了，求解決啊！！！！！！！！！！
uj5u.com熱心網友回復：
請問下，你的解決了嗎？我找了好多檔案也還是不行，解決了麻煩給我傳授下。uj5u.com熱心網友回復：
廣播模式是同一個exchange下的所有queue都收到訊息，你代碼里面只有一個queue，所以只有一條訊息，兩個消費者同時去消費，當然只有一個消費者能收到uj5u.com熱心網友回復：
對啊，不同的消費者要用不同的queue名稱系結，可以讓rabbitmq自己生成名字，也可以自定義。
使用過activemq，rabbitmq的這種概念完全不適應啊，消費者竟然要宣告queue的方式去消費訊息。
uj5u.com熱心網友回復：
這個問題也困擾我好久，我的做法是：生產者宣告一個FANOUT交換機
消費者宣告一個系統產生的隨機佇列系結到這個交換機上，然后往交換機發訊息，只要系結到這個交換機上都能收到訊息，關鍵代碼如下：
生產者：
@Slf4j
@Component
public class RabbitHelper {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 廣播訊息
* @param exchange
* @param message
*/
public void broadcast(String exchange,String message){
rabbitTemplate.convertAndSend(exchange,"",message);
}
}
消費者：
@Slf4j
@Component
public class DownCmdConsumer {
@RabbitListener(bindings = @QueueBinding(
value = @Queue(), //注意這里不要定義佇列名稱,系統會隨機產生
exchange = @Exchange(value = "填寫交換機名稱",type = ExchangeTypes.FANOUT)
)
)
public void process(String payload) {
log.info("receive:{}",payload)
}
}
uj5u.com熱心網友回復：
uj5u.com熱心網友回復：
老哥穩，我看cloud BUS差不多也是這么做的6666
uj5u.com熱心網友回復：
樓主的意思是多個消費者掛在同一個消費佇列中,多個消費者都能接收到.但是因為多個消費者掛在同一個佇列上,會有競爭關系,所以只有一個接收到,其他接收不到(這才是要解決的問題).看了一些博客.有一個方案是采用redis的發布訂閱模式(得到訊息的去通知其他的消費者)轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/50851.html
標籤：Java SE
上一篇：No resolvable bootstrap urls given in bootstrap.servers
下一篇：淘寶api查訂單 怎么獲得權限？