У меня есть 2 конфигурации Spring RabbitMq, одна с использованием RabbitTemplate, другая с использованием JmsTemplate.
Конфигурация с RabbitTemplate:
Класс AmqpMailIntegrationPerfTestConfig :
@Configuration
@ComponentScan(basePackages = {
"com.test.perf.amqp.receiver",
"com.test.perf.amqp.sender"
})
@EnableRabbit
public class AmqpMailIntegrationPerfTestConfig {
@Bean
public DefaultClassMapper classMapper() {
DefaultClassMapper classMapper = new DefaultClassMapper();
Map<String, Class<?>> idClassMapping = new HashMap<>();
idClassMapping.put("mail", MailMessage.class);
classMapper.setIdClassMapping(idClassMapping);
return classMapper;
}
@Bean
public Jackson2JsonMessageConverter jsonMessageConverter() {
Jackson2JsonMessageConverter jsonConverter = new Jackson2JsonMessageConverter();
jsonConverter.setClassMapper(classMapper());
return jsonConverter;
}
@Bean
public RabbitTemplate myRabbitTemplate(ConnectionFactory connectionFactory) {
final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
rabbitTemplate.setMessageConverter(jsonMessageConverter());
return rabbitTemplate;
}
@Bean
public ConnectionFactory createConnectionFactory(){
CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
return connectionFactory;
}
@Bean
Queue queue() {
return new Queue(AmqpMailSenderImpl.QUEUE_NAME, false);
}
@Bean
TopicExchange exchange() {
return new TopicExchange(AmqpMailSenderImpl.TOPIC_EXCHANGE_NAME);
}
@Bean
Binding binding(Queue queue, TopicExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(AmqpMailSenderImpl.ROUTING_KEY);
}
@Bean
public AmqpAdmin amqpAdmin() {
return new RabbitAdmin(createConnectionFactory());
}
@Bean
public SimpleRabbitListenerContainerFactory myRabbitListenerContainerFactory() {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(createConnectionFactory());
factory.setMaxConcurrentConsumers(5);
factory.setMessageConverter(jsonMessageConverter());
return factory;
}
}
Класс AmqpMailSenderPerfImpl в пакете com.test.perf.amqp.sender :
@Component
public class AmqpMailSenderPerfImpl implements MailSender {
public static final String TOPIC_EXCHANGE_NAME = "mails-exchange";
public static final String ROUTING_KEY = "mails";
@Autowired
private RabbitTemplate rabbitTemplate;
@Override
public boolean sendMail(MailMessage message) {
rabbitTemplate.convertAndSend(TOPIC_EXCHANGE_NAME, ROUTING_KEY, message);
return true;
}
}
Класс AmqpMailReceiverPerfImpl в com.test.Комплект perf.amqp.receiver :
@Component
public class AmqpMailReceiverPerfImpl implements ReceivedDatesKeeper {
private Logger logger = LoggerFactory.getLogger(getClass());
private Map<String,Date> datesReceived = new HashMap<String, Date>();
@RabbitListener(containerFactory = "myRabbitListenerContainerFactory", queues = AmqpMailSenderImpl.QUEUE_NAME)
public void receiveMessage(MailMessage message) {
logger.info("------ Received mail! ------\nmessage:" + message.getSubject());
datesReceived.put(message.getSubject(), new Date());
}
public Map<String, Date> getDatesReceived() {
return datesReceived;
}
}
Конфигурация с JmsTemplate:
Класс JmsMailIntegrationPerfTestConfig :
@Configuration
@EnableJms
@ComponentScan(basePackages = {
"com.test.perf.jms.receiver",
"com.test.jms.sender"
})
public class JmsMailIntegrationPerfTestConfig {
@Bean
public MessageConverter jacksonJmsMessageConverter() {
MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
Map<String,Class<?>> typeIdMappings = new HashMap<String,Class<?>>();
typeIdMappings.put("mail", MailMessage.class);
converter.setTypeIdMappings(typeIdMappings);
converter.setTargetType(MessageType.TEXT);
converter.setTypeIdPropertyName("_type");
return converter;
}
@Bean
public ConnectionFactory createConnectionFactory(){
RMQConnectionFactory connectionFactory = new RMQConnectionFactory();
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setVirtualHost("/");
connectionFactory.setHost("localhost");
connectionFactory.setPort(5672);
return connectionFactory;
}
@Bean(name = "myJmsFactory")
public JmsListenerContainerFactory<?> myFactory(ConnectionFactory connectionFactory) {
DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setConcurrency("10-50");
factory.setMessageConverter(jacksonJmsMessageConverter());
return factory;
}
@Bean
public Destination jmsDestination() {
RMQDestination jmsDestination = new RMQDestination();
jmsDestination.setDestinationName("myQueue");
jmsDestination.setAmqp(false);
jmsDestination.setAmqpQueueName("mails");
return jmsDestination;
}
@Bean
public JmsTemplate myJmsTemplate(ConnectionFactory connectionFactory) {
final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
jmsTemplate.setMessageConverter(jacksonJmsMessageConverter());
return jmsTemplate;
}
}
Класс JmsMailSenderImpl в пакете com.test.jms.sender :
@Component
public class JmsMailSenderImpl implements MailSender {
private Logger logger = LoggerFactory.getLogger(getClass());
@Autowired
private JmsTemplate jmsTemplate;
@Override
public boolean sendMail(MailMessage message) {
logger.info("Sending message!");
jmsTemplate.convertAndSend("mailbox", message);
return false;
}
}
Класс JmsMailReceiverPerfImpl в пакете com.test.perf.jms.receiver :
@Component
public class JmsMailReceiverPerfImpl implements ReceivedDatesKeeper {
private Logger logger = LoggerFactory.getLogger(getClass());
private Map<String,Date> datesReceived = new HashMap<String, Date>();
@JmsListener(destination = "mailbox", containerFactory = "myJmsFactory", concurrency = "10")
public void receiveMail(MailMessage message) {
datesReceived.put(message.getSubject(), new Date());
logger.info("Received <" + message.getSubject() + ">");
}
public Map<String, Date> getDatesReceived() {
return datesReceived;
}
}
Я тестирую вышеуказанные конфигурации, запуская 10 потоков и заставляя соответствующие MailSenders отправлять 1000 писем каждое.
Для конфигурации с RabbitTemplate я получаю: * Общая пропускная способностьвремя всех сообщенийes: 3687ms * Время обработки одного сообщения: 817ms
Для конфигурации с JmsTemplate я получаю: * Общая пропускная способность всех сообщений: 41653ms * Время обработки одного сообщения: 67ms
Thisпохоже, указывает на то, что версия с JmsTemplate не работает параллельно или, по крайней мере, не использует ресурсы оптимально.
Кто-нибудь знает, что может быть причиной этого?Я поигрался с разными параметрами транзакций и параллелизма, но безрезультатно.
Нам нужно получить то же время пропускной способности с JmsTemplate, что и с RabbitTemplate, поэтому мы можем использовать JMS в качестве уровня абстракции.