RabbitMQ + SpringAMPQ: получение нескольких очередей в SimpleMessageListenerContainer - PullRequest
Новая
       87

RabbitMQ + SpringAMPQ: получение нескольких очередей в SimpleMessageListenerContainer

0 голосов
/ 14 апреля 2020

В настоящее время я немного изучаю RabbitMQ + SpringAMQ, и я пытался сделать так, чтобы мой SimpleMessageListenerContainer мог читать сообщения, доставленные в две мои очереди, но только одна его получает. Если у вас есть какие-либо дополнительные советы по поводу кода или если есть какой-либо способ сделать это лучше, пожалуйста, дайте мне знать, это было бы неплохо.

РЕДАКТИРОВАТЬ: проблема Код не отправляется сообщение в обе очереди только в одну.

В основном я получаю из двух очередей, одна вставляет что-то в базу данных, другая получает запрос проверки. Я sh позже отправлю ответ на ключ маршрутизации и обменяюсь.

Вот мое приложение. java: 

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 отправляет его моему получателю. java методу receiveMessage, другой вопрос - как я могу прочитать, из какой очереди я получаю сообщение? Это возможно таким образом, или я должен использовать @RabbitListener для указания его на каждое место?

Приемник. java

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 Ответ

0 голосов
/ 14 апреля 2020

В вашем случае, я думаю, что вы должны использовать RabbitListner и передавать имя очереди в качестве параметра следующим образом: 

 @RabbitListener(queues = RabbitTopicConfig.QUEUE_NAME)
    public void consumerDefaultMessage(final Message message){
        LOGGER.info("Recive Message with default configuration {}", message.toString());
    }

И вы должны внедрить rabbitTemplate в класс конфигурации как бин, в в этом случае вы можете использовать много списков, поскольку очереди существуют: 

    @Bean
    public RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(producerMessageConverter());
        return rabbitTemplate;
    }
...