RabbitMQ 开发随笔

消息处理

使用自定义消息转换器

业务之间大多数数据都是以 JSON 的数据格式进行传输的，即生产者服务将 JSON 类型的数据发送到对应的队列，而消费端从队列中接收到的数据类型也是 JSON 类型。为了方便将 POJO 对象转为 JSON 类型的数据来传输，可以使用 Spring 内置的 Jackson2JsonMessageConverter 消息转换器，具体代码如下：

import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMqConfig {

    /**
     * RabbitMQ的管理对象
     *
     * @param connectionFactory
     * @return
     */
    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        return rabbitAdmin;
    }

    /**
     * RabbitMq的消息转换器
     *
     * @return
     */
    @Bean
    public MessageConverter jsonMessageConverter() {
        Jackson2JsonMessageConverter messageConverter = new Jackson2JsonMessageConverter();
        return messageConverter;
    }

    /**
     * RabbitMq的模版
     *
     * @return
     */
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, MessageConverter messageConverter) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        // 设置发送消息时所用的消息转换器
        template.setMessageConverter(messageConverter);
        return template;
    }

    /**
     * RabbitMq的监听容器工厂
     *
     * @return
     */
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory, MessageConverter messageConverter) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        // 设置线程数
        factory.setConcurrentConsumers(3);
        // 最大线程数
        factory.setMaxConcurrentConsumers(10);
        // 设置接收消息时所用的消息转换器
        factory.setMessageConverter(messageConverter);
        return factory;
    }

}

客户端连接

客户端 SSL 连接

RabbitMQ 客户端使用 SSL 证书连接 RabbitMQ 服务器的 Java 示例代码如下：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.GetResponse;

import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManagerFactory;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.security.KeyManagementException;
import java.security.KeyStore;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.security.UnrecoverableKeyException;
import java.security.cert.CertificateException;
import java.util.concurrent.TimeoutException;

public class SslConnectionDemo {

    public static void main(String[] args) throws TimeoutException {
        // 证书路径
        String classpath = SslReceiver.class.getResource("/").getPath();
        // 证书密码
        char[] sslPwd1 = "password1".toCharArray();
        char[] sslPwd2 = "password2".toCharArray();
        // 读取client密钥和rabbitStore证书
        try (InputStream sslCardStream = new FileInputStream(classpath + "ssl/keycert.p12");
             InputStream rabbitStoreStream = new FileInputStream(classpath + "ssl/rabbitStore")) {

            // 加载秘钥
            KeyStore ks = KeyStore.getInstance("PKCS12");
            ks.load(sslCardStream, sslPwd1);
            KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance("SunX509");
            keyManagerFactory.init(ks, sslPwd1);

            // 读取授权证书，只含有服务端的公钥
            KeyStore jks = KeyStore.getInstance("JKS");
            jks.load(rabbitStoreStream, sslPwd2);
            TrustManagerFactory keyStoreManager = TrustManagerFactory.getInstance("SunX509");
            keyStoreManager.init(jks);
            SSLContext context = SSLContext.getInstance("TLSv1.2");
            context.init(keyManagerFactory.getKeyManagers(), keyStoreManager.getTrustManagers(), null);
            ConnectionFactory factory = new ConnectionFactory();
            // RabbitMQ的登录账号与密码
            factory.setUsername("yourUserName");
            factory.setPassword("yourPassword");
            // RabbitMQ的主机地址
            factory.setHost("127.0.0.1");
            // RabbitMQ的SSL端口
            factory.setPort(5671);
            factory.setAutomaticRecoveryEnabled(true);

            // 设置sslContext
            factory.useSslProtocol(context);
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            // 声明RabbitMQ的队列
            channel.queueDeclare("rabbitmq-queue", false, true, true, null);
            channel.basicPublish("", "rabbitmq-queue", null, "Test,Test".getBytes());
            GetResponse chResponse = channel.basicGet("rabbitmq-queue", false);
            if (chResponse == null) {
                System.out.println("No message retrieved");
            } else {
                byte[] body = chResponse.getBody();
                System.out.println("Recieved: " + new String(body));
            }
            channel.close();
            connection.close();
        } catch (KeyStoreException | UnrecoverableKeyException | KeyManagementException | CertificateException | NoSuchAlgorithmException | IOException e) {
            e.printStackTrace();
        }
    }

}

Java 消息队列任务的平滑关闭分析

• https://jaesonchen.iteye.com/blog/2342761