Queues vs Topics and examples with Java, Spring Boot and Apache ActiveMQ

This text is a continuation of an example I did a while ago about producing and consuming a queue. The idea here is to show the difference between using a queue and using a topic.

Difference in delivery model

Consumers in a queue do not receive the same message. In other words, each message goes to one and only one consumer as shown in the example:

Apache ActiveMQ itself has a load balancer that does this.

On the other hand, with a topic it is possible to broadcast, delivering messages to all consumers, like that:

But what is the use of this?

Thinking of a scenario where it is possible to perform more than one action simultaneously this is very useful. I will take a classic example: an online store, this design speaks for itself:

I’m not an expert in this business, just assuming that these tasks are independent in this scenario and a single type of event can initiate them.

Creating a topic in Apache ActiveMQ

In the activemq console, just access the “topics” tab and create a new topic:

Creating a producer

I always start with the properties, so I put this in application.properties:

activemq.broker-url=tcp://localhost:61616
activemq.user=
activemq.password=
activemq.topic-name=topic.person

And I did a configuration class:

@Configuration
@EnableJms
public class JmsConfig {

    @Value( "${activemq.url}" )
    private String brokerUrl;

    @Value( "${activemq.user}" )
    private String user;

    @Value( "${activemq.password}" )
    private String password;

    @Bean
    public ActiveMQConnectionFactory connectionFactory() {
        if ("".equals(user)) {
            return new ActiveMQConnectionFactory(brokerUrl);
        }
        return new ActiveMQConnectionFactory(user, password, brokerUrl);
    }

    @Bean
    public JmsListenerContainerFactory jmsFactoryTopic(ConnectionFactory connectionFactory,
                                                       DefaultJmsListenerContainerFactoryConfigurer configurer) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        factory.setPubSubDomain(true);
        factory.setClientId("produtor");
        factory.setSubscriptionDurable(true);
        return factory;
    }

    @Bean
    public JmsTemplate jmsTemplateTopic() {
        JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory());
        jmsTemplate.setPubSubDomain(true);
        return jmsTemplate;
    }
}

It is important to use the two annotations of this class as they help within the context of spring. It’s also worth noting the line:

factory.setClientId(“producer”);

It creates the ID of that producer when connecting to the topic.
I also created a class to send the messages:

@Component
@RequiredArgsConstructor
public class Producer {

    private final JmsTemplate jmsTemplate;

    @Value("${activemq.topic-name}")
    private String destinationTopic;

    public void send(Person person) throws JMSException {
        Gson gson = new Gson();
        String jsonPerson = gson.toJson(person);

        jmsTemplate.convertAndSend(destinationTopic, jsonPerson);
    }

}

Okay, that’s enough to post messages to the topic.

Creating the consumer

Again I’ll start with the properties:

activemq.broker-url=tcp://localhost:61616
activemq.user=
activemq.password=
activemq.topic-name=topic.person

Now the configuration class:

@Configuration
@EnableJms
public class JmsConfig {

    @Value( "${activemq.broker-url}" )
    private String brokerUrl;

    @Value( "${activemq.user}" )
    private String user;

    @Value( "${activemq.password}" )
    private String password;

    @Bean
    public ActiveMQConnectionFactory connectionFactory() {
        if ("".equals(user)) {
            return new ActiveMQConnectionFactory(brokerUrl);
        }
        return new ActiveMQConnectionFactory(user, password, brokerUrl);
    }

    @Bean
    public JmsListenerContainerFactory jmsListenerContainerFactory(ConnectionFactory connectionFactory,
                                                       DefaultJmsListenerContainerFactoryConfigurer configurer) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        factory.setPubSubDomain(true);
        factory.setClientId("meu-consumer");
        factory.setSubscriptionDurable(true);
        return factory;
    }

    @Bean
    public JmsTemplate jmsTemplate() {
        return new JmsTemplate(connectionFactory());
    }

    @Bean
    public JmsTemplate jmsTemplateTopic() {
        JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory());
        jmsTemplate.setPubSubDomain(true);
        return jmsTemplate;
    }
}

The factory.setClientId(“my-consumer”) line creates an ID when connecting to the topic, and the factory.setSubscriptionDurable(true) line puts consumers in a receiving configuration even if they are offline. This means to say that when they connect they will receive the messages because the broker knows who they are ..

Now the class that actually consumes:

@Slf4j
@RequiredArgsConstructor
@Component
public class ConsumerTopico {

    private final PersonRepository personRepository;

    @JmsListener( destination = "${activemq.topic-name}", subscription = "assinatura", selector = "test=false OR test is null")
    public void listen(String mensagem) {
        log.info(mensagem);
        Person person = new Person();
        try {
            Gson gson = new Gson();
            person = gson.fromJson(mensagem, Person.class);
            personRepository.save(person);
        }catch(Exception e){
            log.error(e.getMessage());
        }
    }

}

The subscription parameter creates a subscription with the topic, so the topic will know which consumers it should send messages to. However, this subscription is only registered after the first connection of this consumer.
Now just create other consumers putting clientId and signature and that’s it, everyone will receive these messages (I tested it here).

Validations in the environment

To check if the consumers have gone right up, just look at the Apache ActiveMQ adm page, usually at http://localhost:8161, just click on “Manage ActiveMQ broker” and then on “Subscribers”. In this one, it is already possible to see the number of consumers connected and whether they have the durable connection that allows them to receive messages even if they spend time offline:

Done.