2. 程式人生 > >(三) RabbitMQ實戰教程(面向Java開發人員)之Spring整合RabbitMQ

(三) RabbitMQ實戰教程(面向Java開發人員)之Spring整合RabbitMQ

阿新 • • 發佈：2018-12-25

Spring整合RabbitMQ

Spring AMQP是對AMQP協議的抽象和封裝，從官方網站上得知它是由兩個專案組成的(spring-amqp和spring-rabbit)。在使用Spring整合RabbitMQ時我們主要關注三個核心介面(MessageListenerContainer、RabbitAdmin以及RabbitTemplate)

RabbitAdmin: 用於宣告交換機 佇列 繫結等
RabbitTemplate: 用於RabbitMQ訊息的傳送和接收
MessageListenerContainer: 監聽容器 為訊息入隊提供非同步處理

使用Spring整合RabbitMQ需要匯入如下依賴

<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
    <version>2.0.2.RELEASE</version>
</dependency>

Spring AMQP手動宣告

1.建立生產者配置類 將RabbitAdmin、RabbitTemplate納入Spring管理

@Configuration
public class
 
 SpringAMQPProducerConfig {
    @Bean
    public ConnectionFactory connectionFactory() {
        com.rabbitmq.client.ConnectionFactory connectionFactory = new com.rabbitmq.client.ConnectionFactory();

        // 配置連線資訊
        connectionFactory.setHost("192.168.56.128");
        connectionFactory.setPort(5672
 
);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("roberto");
        connectionFactory.setPassword("roberto");

        // 網路異常自動連線恢復
        connectionFactory.setAutomaticRecoveryEnabled(true);
        // 每10秒嘗試重試連線一次
        connectionFactory.setNetworkRecoveryInterval(10000);

        // 設定ConnectionFactory屬性資訊
        Map<String, Object> connectionFactoryPropertiesMap = new HashMap();
        connectionFactoryPropertiesMap.put("principal", "RobertoHuang");
        connectionFactoryPropertiesMap.put("description", "RGP訂單系統V2.0");
        connectionFactoryPropertiesMap.put("emailAddress", "[email protected]");
        connectionFactory.setClientProperties(connectionFactoryPropertiesMap);

        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(connectionFactory);
        return cachingConnectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        return new RabbitTemplate(connectionFactory);
    }
}

2.建立生產者啟動類

@ComponentScan(basePackages = "roberto.growth.process.rabbitmq.spring.amqp.manual.declare.producer")
public class ProducerApplication {
    public static void main(String[] args) {
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(ProducerApplication.class);

        RabbitAdmin rabbitAdmin = context.getBean(RabbitAdmin.class);
        RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);

        // 宣告交換機
        rabbitAdmin.declareExchange(new DirectExchange("roberto.order", true, false, new HashMap<>()));

        // 宣告訊息 (訊息體, 訊息屬性)
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        messageProperties.setContentType("UTF-8");
        Message message = new Message("訂單資訊".getBytes(), messageProperties);

        // 釋出訊息 (交換機名, Routing key, 訊息);
        // 釋出訊息還可以使用rabbitTemplate.convertAndSend(); 其支援訊息後置處理
        rabbitTemplate.send("roberto.order", "add", message);
    }
}

3.建立消費者配置類 將RabbitAdmin納入Spring管理，並在MessageListenerContainer類中定義了訊息消費的邏輯

@Configuration
public class SpringAMQPConsumerConfig {
    @Bean
    public ConnectionFactory connectionFactory() {
        com.rabbitmq.client.ConnectionFactory connectionFactory = new com.rabbitmq.client.ConnectionFactory();

        // 配置連線資訊
        connectionFactory.setHost("192.168.56.128");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("roberto");
        connectionFactory.setPassword("roberto");

        // 網路異常自動連線恢復
        connectionFactory.setAutomaticRecoveryEnabled(true);
        // 每10秒嘗試重試連線一次
        connectionFactory.setNetworkRecoveryInterval(10000);

        // 設定ConnectionFactory屬性資訊
        Map<String, Object> connectionFactoryPropertiesMap = new HashMap();
        connectionFactoryPropertiesMap.put("principal", "RobertoHuang");
        connectionFactoryPropertiesMap.put("description", "RGP訂單系統V2.0");
        connectionFactoryPropertiesMap.put("emailAddress", "[email protected]");
        connectionFactory.setClientProperties(connectionFactoryPropertiesMap);

        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(connectionFactory);
        return cachingConnectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    @Bean
    public MessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer messageListenerContainer = new SimpleMessageListenerContainer();
        messageListenerContainer.setConnectionFactory(connectionFactory);
        messageListenerContainer.setQueueNames("roberto.order.add");

        // 設定消費者執行緒數
        messageListenerContainer.setConcurrentConsumers(5);
        // 設定最大消費者執行緒數
        messageListenerContainer.setMaxConcurrentConsumers(10);

        // 設定消費者屬性資訊
        Map<String, Object> argumentMap = new HashMap();
        messageListenerContainer.setConsumerArguments(argumentMap);

        // 設定消費者標籤
        messageListenerContainer.setConsumerTagStrategy(new ConsumerTagStrategy() {
            @Override
            public String createConsumerTag(String s) {
                return "RGP訂單系統ADD處理邏輯消費者";
            }
        });

        // 使用setAutoStartup方法可以手動設定訊息消費時機
        messageListenerContainer.setAutoStartup(false);

        // 使用setAfterReceivePostProcessors方法可以增加訊息後置處理器
        // messageListenerContainer.setAfterReceivePostProcessors();

        messageListenerContainer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                try {
                    System.out.println(new String(message.getBody(), "UTF-8"));
                    System.out.println(message.getMessageProperties());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        return messageListenerContainer;
    }
}

4.建立消費者啟動類

@ComponentScan(basePackages = "roberto.growth.process.rabbitmq.spring.amqp.manual.declare.consumer")
public class ConsumerApplication {
    public static void main(String[] args) {
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(ConsumerApplication.class);

        RabbitAdmin rabbitAdmin = context.getBean(RabbitAdmin.class);
        MessageListenerContainer messageListenerContainer = context.getBean(MessageListenerContainer.class);

        // 宣告佇列 (佇列名", 是否持久化, 是否排他, 是否自動刪除, 佇列屬性);
        rabbitAdmin.declareQueue(new Queue("roberto.order.add", true, false, false, new HashMap<>()));

        // 宣告Direct Exchange (交換機名, 是否持久化, 是否自動刪除, 交換機屬性);
        rabbitAdmin.declareExchange(new DirectExchange("roberto.order", true, false, new HashMap<>()));

        // 將佇列Binding到交換機上 Routing key為add
        rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue("roberto.order.add")).to(new DirectExchange("roberto.order")).with("add"));

        // 開始監聽佇列
        messageListenerContainer.start();
    }
}

5.依次啟動訊息消費者和生產者 控制檯輸出如下

訂單資訊
MessageProperties [headers={}, contentType=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=roberto.order, receivedRoutingKey=add, deliveryTag=1, consumerTag=RGP訂單系統ADD處理邏輯消費者, consumerQueue=roberto.order.add]

上訴程式碼使用Spring整合RabbitMQ，最終實現效果和上一篇部落格介紹RabbitMQ Java Client一致

Spring AMQP自動宣告

在上訴的Demo中我們是手動使用RabbitAdmin對交換機、佇列和繫結進行宣告的，Spring AMQP還提供了自動宣告方式交換機、佇列和繫結。我們可以直接把要自動宣告的元件納入Spring容器中管理即可，自動聲明發生在RabbitMQ第一次連線建立的時候，自動宣告支援單個和批量自動宣告。使用自動宣告需要符合如下條件:

1.需要有連線產生
2.RabbitAdmin必須交由Spring管理，且autoStartup必須為true(預設)
3.如果ConnectionFactory使用的是CachingConnectionFactory，則cacheMode必須要為CacheMode.CHANNEL
4.所有要宣告的元件的shouldDeclare必須為true
5.要宣告的Queue名稱不能以amq.開頭

上訴規則定義在RabbitAdmin的afterPropertiesSet方法中，有興趣的同學可以自行閱讀RabbitAdmin原始碼

1.建立生產者配置類 將RabbitAdmin、RabbitTemplate納入Spring管理

@Configuration
public class SpringAMQPProducerConfig {
    @Bean
    public ConnectionFactory connectionFactory() {
        com.rabbitmq.client.ConnectionFactory connectionFactory = new com.rabbitmq.client.ConnectionFactory();

        // 配置連線資訊
        connectionFactory.setHost("192.168.56.128");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("roberto");
        connectionFactory.setPassword("roberto");

        // 網路異常自動連線恢復
        connectionFactory.setAutomaticRecoveryEnabled(true);
        // 每10秒嘗試重試連線一次
        connectionFactory.setNetworkRecoveryInterval(10000);

        // 設定ConnectionFactory屬性資訊
        Map<String, Object> connectionFactoryPropertiesMap = new HashMap();
        connectionFactoryPropertiesMap.put("principal", "RobertoHuang");
        connectionFactoryPropertiesMap.put("description", "RGP訂單系統V2.0");
        connectionFactoryPropertiesMap.put("emailAddress", "[email protected]");
        connectionFactory.setClientProperties(connectionFactoryPropertiesMap);

        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(connectionFactory);
        return cachingConnectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        return new RabbitTemplate(connectionFactory);
    }
}

2.建立生產者啟動類

@ComponentScan(basePackages = "roberto.growth.process.rabbitmq.spring.amqp.auto.declare.producer")
public class ProducerApplication {
    public static void main(String[] args) {
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(roberto.growth.process.rabbitmq.spring.amqp.manual.declare.producer.ProducerApplication.class);

        RabbitAdmin rabbitAdmin = context.getBean(RabbitAdmin.class);
        RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);

        // 宣告交換機
        rabbitAdmin.declareExchange(new DirectExchange("roberto.order", true, false, new HashMap<>()));

        // 宣告訊息 (訊息體, 訊息屬性)
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        messageProperties.setContentType("UTF-8");
        Message message = new Message("訂單資訊".getBytes(), messageProperties);

        // 釋出訊息 (交換機名, Routing key, 訊息);
        // 釋出訊息還可以使用rabbitTemplate.convertAndSend(); 其支援訊息後置處理
        rabbitTemplate.send("roberto.order", "add", message);
    }
}

3.建立消費者配置類 將RabbitAdmin納入Spring管理，並在MessageListenerContainer類中定義了訊息消費的邏輯，並且在該配置類中宣告交換機，佇列，繫結

@Configuration
public class SpringAMQPConsumerConfig {
    @Bean
    public ConnectionFactory connectionFactory() {
        com.rabbitmq.client.ConnectionFactory connectionFactory = new com.rabbitmq.client.ConnectionFactory();

        // 配置連線資訊
        connectionFactory.setHost("192.168.56.128");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("roberto");
        connectionFactory.setPassword("roberto");

        // 網路異常自動連線恢復
        connectionFactory.setAutomaticRecoveryEnabled(true);
        // 每10秒嘗試重試連線一次
        connectionFactory.setNetworkRecoveryInterval(10000);

        // 設定ConnectionFactory屬性資訊
        Map<String, Object> connectionFactoryPropertiesMap = new HashMap();
        connectionFactoryPropertiesMap.put("principal", "RobertoHuang");
        connectionFactoryPropertiesMap.put("description", "RGP訂單系統V2.0");
        connectionFactoryPropertiesMap.put("emailAddress", "[email protected]");
        connectionFactory.setClientProperties(connectionFactoryPropertiesMap);

        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(connectionFactory);
        return cachingConnectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    @Bean
    // 自動宣告交換機
    // 如果要一次性宣告多個 使用public List<Exchange> listExchange()即可
    public Exchange exchange() {
        return new DirectExchange("roberto.order", true, false, new HashMap<>());
    }

    @Bean
    // 自動宣告佇列
    // 如果要一次性宣告多個 使用public List<Queue> listQueue()即可
    public Queue queue() {
        return new Queue("roberto.order.add", true, false, false, new HashMap<>());
    }

    @Bean
    // 自動宣告繫結
    // 如果要一次性宣告多個 使用public List<Binding> listBinding()即可
    public Binding binding() {
        return new Binding("roberto.order.add", Binding.DestinationType.QUEUE, "roberto.order", "add", new HashMap<>());
    }

    @Bean
    public MessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer messageListenerContainer = new SimpleMessageListenerContainer();
        messageListenerContainer.setConnectionFactory(connectionFactory);
        messageListenerContainer.setQueueNames("roberto.order.add");

        // 設定消費者執行緒數
        messageListenerContainer.setConcurrentConsumers(5);
        // 設定最大消費者執行緒數
        messageListenerContainer.setMaxConcurrentConsumers(10);

        // 設定消費者屬性資訊
        Map<String, Object> argumentMap = new HashMap();
        messageListenerContainer.setConsumerArguments(argumentMap);

        // 設定消費者標籤
        messageListenerContainer.setConsumerTagStrategy(new ConsumerTagStrategy() {
            @Override
            public String createConsumerTag(String s) {
                return "RGP訂單系統ADD處理邏輯消費者";
            }
        });

        // 使用setAutoStartup方法可以手動設定訊息消費時機
        messageListenerContainer.setAutoStartup(true);

        // 使用setAfterReceivePostProcessors方法可以增加訊息後置處理器
        // messageListenerContainer.setAfterReceivePostProcessors();

        messageListenerContainer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                try {
                    System.out.println(new String(message.getBody(), "UTF-8"));
                    System.out.println(message.getMessageProperties());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        return messageListenerContainer;
    }
}

4.建立消費者啟動類

@ComponentScan(basePackages = "roberto.growth.process.rabbitmq.spring.amqp.auto.declare.consumer")
public class ConsumerApplication {
    public static void main(String[] args) {
        new AnnotationConfigApplicationContext(ConsumerApplication.class);
    }
}

5.依次啟動訊息消費者和生產者 控制檯輸出如下

訂單資訊
MessageProperties [headers={}, contentType=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=roberto.order, receivedRoutingKey=add, deliveryTag=1, consumerTag=RGP訂單系統ADD處理邏輯消費者, consumerQueue=roberto.order.add]

該程式碼與上訴程式碼實現的效果一致，只是將交換機，佇列，繫結進行了自動宣告

MessageListenerAdapte

以上兩個Demo在消費訊息處理邏輯時往MessageListenerContainer中傳遞了MessageListener，但是我們有時候已經寫好了消費邏輯對應的類，我們不希望它去擴充套件MessageListener/ChannelAwareMessageListener，因為這麼做的話意味著我們需要改變現有程式碼。Spring AMQP提供了訊息處理器介面卡的功能，它可以把一個純POJO類適配成一個可以處理訊息的處理器，預設處理訊息的方法為handleMessage，可以通過setDefaultListenerMethod方法進行修改

1.建立生產者配置類

@Configuration
public class SpringAMQPProducerConfig {
    @Bean
    public ConnectionFactory connectionFactory() {
        com.rabbitmq.client.ConnectionFactory connectionFactory = new com.rabbitmq.client.ConnectionFactory();

        connectionFactory.setHost("192.168.56.128");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("roberto");
        connectionFactory.setPassword("roberto");

        connectionFactory.setAutomaticRecoveryEnabled(true);
        connectionFactory.setNetworkRecoveryInterval(10000);

        Map<String, Object> connectionFactoryPropertiesMap = new HashMap();
        connectionFactoryPropertiesMap.put("principal", "RobertoHuang");
        connectionFactoryPropertiesMap.put("description", "RGP訂單系統V2.0");
        connectionFactoryPropertiesMap.put("emailAddress", "[email protected]");
        connectionFactory.setClientProperties(connectionFactoryPropertiesMap);

        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(connectionFactory);
        return cachingConnectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        return new RabbitTemplate(connectionFactory);
    }
}

2.建立生產者啟動類

@ComponentScan(basePackages = "roberto.growth.process.rabbitmq.spring.amqp.message.listener.adapte.producer")
public class ProducerApplication {
    public static void main(String[] args) {
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(ProducerApplication.class);

        RabbitAdmin rabbitAdmin = context.getBean(RabbitAdmin.class);
        RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);

        rabbitAdmin.declareExchange(new DirectExchange("roberto.order", true, false, new HashMap<>()));

        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        messageProperties.setContentType("UTF-8");
        Message message = new Message("訂單資訊".getBytes(), messageProperties);

        rabbitTemplate.send("roberto.order", "add", message);
    }
}

3.建立消費者訊息處理器類，它可是是純POJO類

public class MessageHandle {
    public void add(byte[] message){
        try {
            System.out.println(new String(message,"UTF-8"));
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
    }
}

4.建立消費者配置類 配置自定義訊息處理器(將roberto.order.add佇列使用自定義訊息處理類的add方法進行處理)

@Configuration
public class SpringAMQPConsumerConfig {
    @Bean
    public ConnectionFactory connectionFactory() {
        com.rabbitmq.client.ConnectionFactory connectionFactory = new com.rabbitmq.client.ConnectionFactory();

        connectionFactory.setHost("192.168.56.128");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("roberto");
        connectionFactory.setPassword("roberto");

        connectionFactory.setAutomaticRecoveryEnabled(true);
        connectionFactory.setNetworkRecoveryInterval(10000);

        Map<String, Object> connectionFactoryPropertiesMap = new HashMap();
        connectionFactoryPropertiesMap.put("principal", "RobertoHuang");
        connectionFactoryPropertiesMap.put("description", "RGP訂單系統V2.0");
        connectionFactoryPropertiesMap.put("emailAddress", "[email protected]");
        connectionFactory.setClientProperties(connectionFactoryPropertiesMap);

        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(connectionFactory);
        return cachingConnectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    @Bean
    public Exchange exchange() {
        return new DirectExchange("roberto.order", true, false, new HashMap<>());
    }

    @Bean
    public Queue queue() {
        return new Queue("roberto.order.add", true, false, false, new HashMap<>());
    }

    @Bean
    public Binding binding() {
        return new Binding("roberto.order.add", Binding.DestinationType.QUEUE, "roberto.order", "add", new HashMap<>());
    }

    @Bean
    public MessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer messageListenerContainer = new SimpleMessageListenerContainer();
        messageListenerContainer.setConnectionFactory(connectionFactory);
        messageListenerContainer.setQueueNames("roberto.order.add");

        messageListenerContainer.setConcurrentConsumers(5);
        messageListenerContainer.setMaxConcurrentConsumers(10);

        Map<String, Object> argumentMap = new HashMap();
        messageListenerContainer.setConsumerArguments(argumentMap);

        messageListenerContainer.setConsumerTagStrategy(new ConsumerTagStrategy() {
            @Override
            public String createConsumerTag(String s) {
                return "RGP訂單系統ADD處理邏輯消費者";
            }
        });

        messageListenerContainer.setAutoStartup(true);

        // 新建訊息處理器介面卡
        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(new MessageHandle());
        // 設定預設處理訊息方法
        messageListenerAdapter.setDefaultListenerMethod("handleMessage");
        Map<String, String> queueOrTagToMethodName = new HashMap<>();
        // 將roberto.order.add佇列的訊息 使用add方法進行處理
        queueOrTagToMethodName.put("roberto.order.add","add");
        messageListenerAdapter.setQueueOrTagToMethodName(queueOrTagToMethodName);
        messageListenerContainer.setMessageListener(messageListenerAdapter);

        return messageListenerContainer;
    }
}

5.建立消費者啟動類

@ComponentScan(basePackages = "roberto.growth.process.rabbitmq.spring.amqp.message.listener.adapte.consumer")
public class ConsumerApplication {
    public static void main(String[] args) {
        new AnnotationConfigApplicationContext(ConsumerApplication.class);
    }
}

6.依次啟動訊息消費者和生產者 控制檯輸出如下

訂單資訊

如上Demo說明我們可以將一個純POJO類定義為訊息處理器，並且不用去擴充套件MessageListener/ChannelAwareMessageListener介面，關於自定義處理器方法的引數預設情況下為byte[]型別，這是由Spring AMQP預設訊息轉換器(SimpleMessageConverter)決定的，接下來我們將介紹Spring AMQP的訊息轉換器功能

Spring AMQP訊息轉換器

在上訴例子中我們定義的add(byte[] message)方法的引數是一個位元組陣列，但是有時候我們往RabbitMQ中傳送的是一個JSON物件，我們希望在處理訊息的時候它已經自動幫我們轉為JAVA物件；又或者我們往RabbitMQ中傳送的是一張圖片或其他格式的檔案，我們希望在處理訊息的時候它已經自動幫我們轉成檔案格式，我們可以手動設定MessageConverter來實現如上需求，如果未設定MessageConverter則使用Spring AMQP預設提供的SimpleMessageConverter

以下例子使用MessageConverter實現了當生產者往RabbitMQ傳送不同型別的資料的時候，使用MessageHandle不同的方法進行處理，需要注意的是當生產者在傳送JSON資料的時候，需要制定這個JSON是哪個物件，用於Spring AMQP轉換，規則如下

當傳送普通物件的JSON資料時，需要在訊息的header中增加一個__TypeId__的屬性告知消費者是哪個物件

當傳送List集合物件的JSON資料時，需要在訊息的header中將__TypeId__指定為java.util.List，並且需要額外指定屬性__ContentTypeId__使用者告知消費者List集合中的物件型別

當傳送Map集合物件的JSON資料時，需要在訊息的header中將__TypeId__指定為java.util.Map，並且需要額外指定屬性__KeyTypeId__用於告知客戶端Map中key的型別，__ContentTypeId__用於告知客戶端Map中Value的型別

1.建立訂單實體類

public class Order {
    /**
     * 訂單編號
     **/
    private String orderId;

    /**
     * 訂單金額
     **/
    private BigDecimal orderAmount;

    public Order() {

    }

    public Order(String orderId, BigDecimal orderAmount) {
        this.orderId = orderId;
        this.orderAmount = orderAmount;
    }

    public String getOrderId() {
        return orderId;
    }

    public void setOrderId(String orderId) {
        this.orderId = orderId;
    }

    public BigDecimal getOrderAmount() {
        return orderAmount;
    }

    public void setOrderAmount(BigDecimal orderAmount) {
        this.orderAmount = orderAmount;
    }
}

2.自定義檔案訊息轉換器

public class FileMessageConverter implements MessageConverter {
    @Override
    public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
        return null;
    }

    @Override
    public Object fromMessage(Message message) throws MessageConversionException {
        String extName = (String) message.getMessageProperties().getHeaders().get("_extName");
        byte[] bytes = message.getBody();
        String fileName = UUID.randomUUID().toString();
        String filePath = System.getProperty("java.io.tmpdir") + fileName + "." + extName;
        File tempFile = new File(filePath);
        try {
            FileCopyUtils.copy(bytes, tempFile);
        } catch (IOException e) {
            throw new MessageConversionException("FileMessageConverter訊息轉換失敗", e);
        }

        return tempFile;
    }
}

3.自定義字串訊息轉換器

public class StringMessageConverter implements MessageConverter{
    @Override
    public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
        return null;
    }

    @Override
    public Object fromMessage(Message message) throws MessageConversionException {
        try {
            return new String(message.getBody(), "UTF-8");
        } catch (UnsupportedEncodingException e) {
            throw new MessageConversionException("StringMessageConverter轉換失敗", e);
        }
    }
}

4.自定義訊息處理器類

public class MessageHandle {
    public void add(byte[] body) {
        System.out.println("----------byte[]方法進行處理----------");
        System.out.println("body");
    }

    public void add(String message) {
        System.out.println("----------String方法進行處理----------");
        System.out.println(message);
    }

    public void add(File file) {
        System.out.println("----------File方法進行處理----------");
        System.out.println(file.length());
        System.out.println(file.getName());
        System.out.println(file.getAbsolutePath());
    }

    public void add(Order order) {
        System.out.println("----------Order方法進行處理----------");
        System.out.println(order.getOrderId() + "---" + order.getOrderAmount());
    }

    public void add(List<Order> orderList) {
        System.out.println("----------List<Order>方法進行處理----------");
        System.out.println(orderList.size());
        for (Order order : orderList) {
            System.out.println(order.getOrderId() + "---" + order.getOrderAmount());
        }
    }

    public void add(Map<String, Order> orderMap) {
        System.out.println("----------Map<String, Order>方法進行處理----------");
        for (Map.Entry<String, Order> entry : orderMap.entrySet()) {
            System.out.println(entry.getKey());
            System.out.println(entry.getValue().getOrderId() + "---" + entry.getValue().getOrderAmount());
        }
    }
}

5.建立生產者配置類

@Configuration
public class SpringAMQPProducerConfig {
    @Bean
    public ConnectionFactory connectionFactory() {
        com.rabbitmq.client.ConnectionFactory connectionFactory = new com.rabbitmq.client.ConnectionFactory();

        connectionFactory.setHost("192.168.56.128");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("roberto");
        connectionFactory.setPassword("roberto");

        connectionFactory.setAutomaticRecoveryEnabled(true);
        connectionFactory.setNetworkRecoveryInterval(10000);

        Map<String, Object> connectionFactoryPropertiesMap = new HashMap();
        connectionFactoryPropertiesMap.put("principal", "RobertoHuang");
        connectionFactoryPropertiesMap.put("description", "RGP訂單系統V2.0");
        connectionFactoryPropertiesMap.put("emailAddress", "[email protected]");
        connectionFactory.setClientProperties(connectionFactoryPropertiesMap);

        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(connectionFactory);
        return cachingConnectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        return new RabbitTemplate(connectionFactory);
    }
}

6.建立生產者啟動類 傳送多條不同訊息型別訊息

@ComponentScan(basePackages = "roberto.growth.process.rabbitmq.spring.amqp.message.converter.producer")
public class ProducerApplication {
    public static void main(String[] args) throws Exception {
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(ProducerApplication.class);

        RabbitAdmin rabbitAdmin = context.getBean(RabbitAdmin.class);
        RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);

        rabbitAdmin.declareExchange(new DirectExchange("roberto.order", true, false, new HashMap<>()));

        // 傳送字串
        sendString(rabbitTemplate);
        // 傳送當個物件JSON
        sendSingle(rabbitTemplate);
        // 傳送List集合JSON
        sendList(rabbitTemplate);
        // 傳送Map集合JSON
        sendMap(rabbitTemplate);
        // 傳送圖片
        sendImage(rabbitTemplate);
    }

    public static void sendString(RabbitTemplate rabbitTemplate) {
        // 宣告訊息 (訊息體, 訊息屬性)
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        messageProperties.setContentType("text/plain");
        Message message = new Message("訂單訊息".getBytes(), messageProperties);
        rabbitTemplate.send("roberto.order", "add", message);
    }

    public static void sendSingle(RabbitTemplate rabbitTemplate) throws Exception {
        Order order = new Order("OD0000001", new BigDecimal(888888.888888));
        ObjectMapper objectMapper = new ObjectMapper();

        // 宣告訊息 (訊息體, 訊息屬性)
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.getHeaders().put("__TypeId__", "order");
        messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        messageProperties.setContentType("application/json");
        Message message = new Message(objectMapper.writeValueAsString(order).getBytes(), messageProperties);

        rabbitTemplate.send("roberto.order", "add", message);
    }

    public static void sendList(RabbitTemplate rabbitTemplate) throws Exception {
        Order order = new Order("OD0000001", new BigDecimal(888888.888888));
        Order order2 = new Order("OD0000002", new BigDecimal(888888.888888));
        List<Order> orderList = Arrays.asList(order, order2);

        ObjectMapper objectMapper = new ObjectMapper();

        // 宣告訊息 (訊息體, 訊息屬性)
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.getHeaders().put("__TypeId__", "java.util.List");
        messageProperties.getHeaders().put("__ContentTypeId__", "order");
        messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        messageProperties.setContentType("application/json"

相關推薦

(三) RabbitMQ實戰教程(面向Java開發人員)之Spring整合RabbitMQ

Spring整合RabbitMQ Spring AMQP是對AMQP協議的抽象和封裝，從官方網站上得知它是由兩個專案組成的(spring-amqp和spring-rabbit)。在使用Spring整合RabbitMQ時我們主要關注三個核心介面(MessageL

(九) RabbitMQ實戰教程(面向Java開發人員)之SpringBoot整合RabbitMQ

SpringBoot整合RabbitMQ 使用SpringBoot整合RabbitMQ非常簡單，它極大程度的簡化了開發成本，使用SpringBoot整合RabbitMQ需匯入如下依賴 <parent> <groupId>o

(四) RabbitMQ實戰教程(面向Java開發人員)之@RabbitListener訊息消費

使用RabbitListener註解進行訊息消費 在前一篇部落格中我們往MessageListenerContainer設定了MessageListener進行訊息的消費，本篇部落格將介紹一種更為簡單的訊息消費方式:使用@RabbitListener註解方式。

(七) RabbitMQ實戰教程(面向Java開發人員)之RabbitMQ常用屬性詳解

RabbitMQ常用屬性詳解 Alternate Exchange Alternate Exchange簡稱AE，當訊息不能被正確路由時，如果交換機設定了AE則訊息會被投遞到AE中，如果存在AE鏈則會按此繼續投遞，直到訊息被正確路由或AE鏈結束訊息被

(八) RabbitMQ實戰教程(面向Java開發人員)之RabbitMQ實現非同步RPC

RabbitMQ實現非同步RPC RabbitMQ Java Client 服務端步驟: 1.服務端監聽一個佇列，監聽客戶端傳送過來的訊息 2.收到訊息之後呼叫RPC服務得到呼叫結果 3.從訊息屬性中獲取reply_to，correlation_i

(二) RabbitMQ實戰教程(面向Java開發人員)之Rabbit Java Client

RabbitMQ Java Client 在介紹完RabbitMQ基本概念後，我們使用JAVA程式碼來模擬一套生產者和消費者的模型，Talk is cheap 直接上程式碼了。使用Java Client整合RabbitMQ需要在pom.xml中匯入如下依賴

(六) RabbitMQ實戰教程(面向Java開發人員)之RabbitMQ訊息的可靠性

訊息可靠性 在專案中使用RabbitMQ時，我們可能會遇到這樣的問題:如一個訂單系統當用戶付款成功時我們往訊息中介軟體新增一條記錄期望訊息消費者修改訂單狀態，但是最終實際訂單狀態並沒有被修改成功。遇到這種問題我們排查的思路如下: 1.訊息是否已經成功傳送

(五) RabbitMQ實戰教程(面向Java開發人員)之RabbitMQ異常處理

RabbitMQ異常處理 使用JAVA客戶端整合RabbitMQ進行的許多操作都會丟擲異常，我們可以自定義異常處理器進行處理，比如我們希望在RabbitMQ消費訊息失敗時記錄一條日誌，又或者在訊息消費失敗時傳送一則通知等操作 RabbitMQ Jav

(一) RabbitMQ實戰教程(面向Java開發人員)之RabbitMQ相關概念介紹

前言 因為專案組需要對RabbitMQ的使用進行優化，所以系統化的學習了RabbitMQ，寫下本系列部落格的目的是希望能幫助到後續使用RabbitMQ的同學少走彎路，在閱讀本部落格前我希望您已經對RabbitMQ有基本的瞭解。本篇部落格主要面向JAVA開發人員

關於pytho面向過程開發人員三步轉oo的補充，再加一步，四步走戰略

這樣的 存儲 一個人 類屬性 bsp 哪些 不用 back style 1、之前寫的這個很久了 裏面提到的是三步走： 模塊和類的轉換規則是： 1、模塊級降為類 2、全局變量改成實例屬性，全局的不會被改變的變量類似於那種const的，可以寫成類屬性（減少點

IntelliJ IDEA詳細配置和使用教程(適用於Java開發人員)

前言 正所謂工欲善其事必先利其器，對開發人員而言若想提高編碼效率，一款高效的開發工具是必不可少的，相信看到該部落格的朋友們都已經對IntelliJ IDEA有所瞭解了，所以此處就不對IntelliJ IDEA進行介紹，而是直接闡述在實際開發中應用場景。博主使用

三年經驗java開發人員面試總結。

作者07年畢業，畢業後一直從事於javaEE、ssh框架的B/S架構的軟體開發工作，算是三年多相關的工作經驗，呆了兩個公司。最近離職重新找工作，面試了n家公司，頗有感慨，希望寫出面試經歷與大家分享。 首先，面試前的準備。 準備應包括幾個方面，一是簡歷。簡歷不能太長，但畢竟

三年經驗java開發人員面試總結

作者07年畢業，畢業後一直從事於javaEE、ssh框架的B/S架構的軟體開發工作，算是三年多相關的工作經驗，呆了兩個公司。最近離職重新找工作，面試了n家公司，頗有感慨，希望寫出面試經歷與大家分享。 首先，面試前的準備

給Java開發人員的Play Framework(2.4)介紹 Part1：Play的優缺點以及適用場景

dead 跟著 框架 sta web 錯誤 gist 解壓 環遊世界 1. 關於這篇系列這篇系列不是Play框架的Hello World，由於這樣的文章網上已經有非常多。這篇系列會首先結合實際代碼介紹Play的特點以及適用場景。然後會有幾篇文章介紹Play與Spring

我是一名java開發人員,ibernae以四川成都春熙路發生

。。 原創文章 django 本地 b- 發展 動畫 統一 之間 谷歌在安卓P版本中已經提供了統一的適配方案,可是在安卓O版本上如何適配呢?本文將詳細介紹華為安卓O版本劉海屏適配方案。使用華為提供的劉海屏SDK進 最近一個買房經驗分享在 GiHub 上火了,不到四天的時間迅

Java基礎教程(2)--Java開發環境

特殊 .html 完成 能夠 println 個數 過程 intel 免費試用 一.JVM、JRE和JDK的概念 ??對於初學者來說，這三個術語出現的頻率很高，而且有關這它們的問題在面試題中也會經常出現。因此，理解它們的定義、區別和聯系就顯得尤為重要。在學習這幾個專業術語之

RabbitMQ實戰教程-鍾林森-專題視訊課程

RabbitMQ實戰教程—648人已學習 課程介紹         本課程適用於RabbitMQ實戰者以及SpringBoot整合RabbitMQ實戰需求者。課程將首先帶領大家拜讀RabbitMQ官

優秀的Java開發人員只能待在底層嗎？

很多人做Java開發2，3年後，都會感覺自己遇到瓶頸。什麼都會又什麼都不會，如何改變困境，為什麼很多人寫了7,8年還是一個碼農，工作中太多被動是因為不懂底層原理。公司的工作節奏又比較快，難有機會學習架構原理，也沒人教，所以這個時候，學習架構原理，擴充套件思維，對自己以後職業生涯尤為重要。 同樣公

寫給伺服器端Java開發人員的Kotlin簡介

Kotlin簡介 JetBrains有一個明確的目標：讓Kotlin成為一種多平臺語言，並提供100%的Java互操作性。Kotlin最近的成功和成熟水平為它進入伺服器端提供了一個很好的機會。 選擇Kotlin的理由 許多語言都試圖成為更好的Java。Kotlin在語言和生態系統方面做得都很好。成為更

Spring Boot 針對 Java 開發人員的安裝指南

Spring Boot 可以使用經典的開發工具或者使用安裝的命令列工具。不管使用何種方式，你都需要確定你的 Java 版本為 Java SDK v1.8 或者更高的版本。在你開始安裝之前，你需要確定你當前安裝的 Java 版本滿足系統執行的需要。 你可以使用下面的命令進行檢視