如何使用@Payload将发布者消息绑定到我的Receiver中的自定义类

问题描述 投票:-1回答:1

在我的应用程序中,我将从FileProcess服务之一发布消息(它将处理CSV文件并将其转换为CSVPojo并使用RabbitTemplate发布到队列。

rabbitTemplate.convertAndSend("spring-boot-rabbitmq-BulkSolve.async_BulkSolve_Msg", "BulkSolve_GeneralrequestQueue", pojo);

我有另一个服务BusinessProcess服务,必须侦听此队列并获取消息并对这些消息执行一些业务流程。为此,我们打算使用SpringBatch执行此操作,因此我创建了一个将侦听队列和进程的作业。该作业的触发点如下。

@EnableRabbit

public class Eventscheduler {

@Autowired
Job csvJob;

@Autowired
private JobLauncher jobLauncher;

//@Scheduled(cron="0 */2 * ? * *")
@RabbitListener(queues ="BulkSolve_GeneralrequestQueue")
public void trigger(){
    Reader.batchstatus=false;
    Map<String,JobParameter> maps= new HashMap<String,JobParameter>();

    maps.put("time", new JobParameter(System.currentTimeMillis()));
    JobParameters jobParameters = new JobParameters(maps);
    JobExecution execution=null;
    try {
        //JobLauncher jobLauncher = new JobLauncher();
         execution=jobLauncher.run(csvJob, jobParameters);
    } catch (JobExecutionAlreadyRunningException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    } catch (JobRestartException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    } catch (JobInstanceAlreadyCompleteException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    } catch (JobParametersInvalidException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    System.out.println("JOB Executed:" + execution.getStatus());

}

}

所以当有一个msg发布到这个队列时,我的工作就会触发。在我的工作触发工作后,我的读者会遇到异常。在读者中我得到的是异常。

org.springframework.amqp.support.converter.MessageConversionException: failed to resolve class name [com.comcast.FileProcess.Pojo.CSVPojo]

下面是我用来读取消息作为接收者的Reader类。

@Component
public class Reader extends AmqpItemReader<List<RequestPojo>>{

    @Autowired
     @Qualifier("rabbitTemplate")
     private RabbitTemplate rabbitTemplate;

    public static boolean batchstatus;

    private List<RequestPojo> reqList = new ArrayList<RequestPojo>();

/*  @Autowired
    private SimpleMessageListenerContainer messagelistener;*/

    public Reader(AmqpTemplate rabbitTemplate) {
        super(rabbitTemplate);
        // TODO Auto-generated constructor stub
    }

    List<RequestPojo> msgList = new ArrayList<RequestPojo>();
    @Override
    @SuppressWarnings("unchecked")
    public List<RequestPojo> read() {



        if(!batchstatus){
        RequestPojo msg=(RequestPojo)rabbitTemplate.receiveAndConvert("BulkSolve_GeneralrequestQueue");

            //return (List<RequestPojo>) rabbitTemplate.receive();
        System.out.println("I am inside Reader" );
        msgList.add((RequestPojo) msg);

        //Object result = rabbitTemplate.receiveAndConvert();

        batchstatus=true;

        return  msgList;
        }
        return null;
    }
} 

消费者在这里使用来自发布者的pacakge名称来获取Pojo类。我能够使用@Payload消费消息下面是我使用成功消费消息的代码(下面是该代码)但我想使用RabbitTemplate.receiveAndConvert(“QueueName”)消费消息,我在我的Reader类中显示。

/*Below code sucesfully consumed messages from receiver side using @Payload*/
@RabbitHandler
    @RabbitListener(containerFactory = "simpleMessageListenerContainerFactory", queues ="BulkSolve_GeneralrequestQueue")
    public void subscribeToRequestQueue(@Payload RequestPojo sampleRequestMessage, Message message) throws InterruptedException {

System.out.println(sampleRequestMessage.toString());
}

任何人都可以帮助解决我的错误,使用RabbitTemplate.receiveAndConvert(“QueueName”)从Receiver中使用已发布的消息

根据你的建议,我已经对Jackson2JsonMessageConverter进行了一些配置更改,以便将消息绑定到我的自定义类RequestPojo,如下所示,但它仍然没有将消息绑定到我的自定义类。你能告诉我我在这里做错了什么,并建议我做些什么才能让它发挥作用。

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

        @Bean
        public MessageConverter jsonMessageConverter() {
            return jsonCustomMessageConverter();
        }

        @Bean
        public Jackson2JsonMessageConverter jsonCustomMessageConverter() {
            Jackson2JsonMessageConverter jsonConverter = new Jackson2JsonMessageConverter();
            jsonConverter.setClassMapper(classMapper());
            return jsonConverter;

        }
        @Bean
        public DefaultClassMapper classMapper() {
            DefaultClassMapper classMapper = new DefaultClassMapper();
            Map<String, Class<?>> idClassMapping = new HashMap<String, Class<?>>();
            idClassMapping.put("RequestPojo", RequestPojo.class);
           // idClassMapping.put("bar", Bar.class);
            classMapper.setIdClassMapping(idClassMapping);
            return classMapper;
        }

根据您的建议更改但是低于错误。

      at java.lang.Thread.run(Thread.java:748) [na:1.8.0_131]
Caused by: org.springframework.amqp.support.converter.MessageConversionException: Cannot handle message
        ... 15 common frames omitted
Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [[B] to [com.comcast.BusinessProcess.Pojos.RequestPojo] for GenericMessage [payload=byte[230], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedRoutingKey=BulkSolve_SummaryrequestQueue, amqp_contentEncoding=UTF-8, amqp_receivedExchange=spring-boot-rabbitmq-BulkSolve_summary.async_BulkSolve_Msg, amqp_deliveryTag=1, amqp_consumerQueue=BulkSolve_SummaryrequestQueue, amqp_redelivered=false, id=d79db57c-3cd4-d104-a343-9373215400b8, amqp_consumerTag=amq.ctag-sYwuWA5pmN07gnEUTO-p6A, contentType=application/json, __TypeId__=com.comcast.FileProcess.Pojo.CSVPojo, timestamp=1535661077865}]
        at org.springframework.messaging.handler.annotation.support.PayloadArgumentResolver.resolveArgument(PayloadArgumentResolver.java:142) ~[spring-messaging-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
        at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:112) ~[spring-messaging-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
        at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:135) ~[spring-messaging-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
        at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:107) ~[spring-messaging-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
        at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:49) ~[spring-rabbit-1.7.4.RELEASE.jar!/:na]
        at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:126) ~[spring-rabbit-1.7.4.RELEASE.jar!/:na]
        ... 14 common frames omitted

下面是我的RabbitConfiguration类。

@Configuration("asyncRPCConfig")
@EnableScheduling
@EnableRabbit
public class RabbitMqConfiguration {

public static String replyQueue;

    public static String directExchange;

    public static String requestRoutingKey;

    public static String replyRoutingKey;

    //public static final int threads=3;

    /*@Bean
    public ExecutorService executorService(){
        return Executors.newFixedThreadPool(threads);
    }*/

    /*@Bean
    public CsvPublisher csvPublisher(){
        return new CsvPublisher();
    }
    @Bean
    public ExcelPublisher excelPublisher(){

        return new ExcelPublisher();
    }*/
    /*@Bean
    public GeneralQueuePublisher generalQueuePublisher(){
        return new GeneralQueuePublisher();
    }
    */
    /*@Bean
    public SummaryQueuePublisher summaryQueuePublisher(){
        return new SummaryQueuePublisher();
    }*/
    /*@Bean
    public Subscriber subscriber(){
        return new Subscriber();
    }*/
    /*@Bean
    public Subscriber1 subscriber1(){
        return new Subscriber1();
    }
    @Bean
    public Subscriber2 subscriber2(){
        return new Subscriber2();
    }
    @Bean
    public RestClient restClient(){
        return new RestClient();
    }*/
    /*@Bean
    public SubscriberGeneralQueue1 SubscriberGeneralQueue1(){
        return new SubscriberGeneralQueue1();
    }*/
    /*@Bean
    public SubscriberSummaryQueue1 SubscriberSummaryQueue1(){
        return new SubscriberSummaryQueue1();
    }*/
    @Bean
    public Eventscheduler Eventscheduler(){
        return new Eventscheduler();
    }
     @Bean
        public Executor taskExecutor() {
            return Executors.newCachedThreadPool();
        }

        @Bean
        public SimpleRabbitListenerContainerFactory simpleMessageListenerContainerFactory(ConnectionFactory connectionFactory,
                                                                                          SimpleRabbitListenerContainerFactoryConfigurer configurer) {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setTaskExecutor(taskExecutor());
            configurer.configure(factory, connectionFactory);
            return factory;
        }

       /* @Bean
        public SimpleRabbitListenerContainerFactory simpleMessageListenerContainerFactory_Summary(ConnectionFactory connectionFactory,
                                                                                          SimpleRabbitListenerContainerFactoryConfigurer configurer) {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setTaskExecutor(taskExecutor());
            configurer.configure(factory, connectionFactory);
            return factory;
        }*/



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

        @Bean
        public MessageConverter jsonMessageConverter() {
            return jsonCustomMessageConverter();
        }

        @Bean
        public Jackson2JsonMessageConverter jsonCustomMessageConverter() {
            Jackson2JsonMessageConverter jsonConverter = new Jackson2JsonMessageConverter();
            jsonConverter.setClassMapper(classMapper());
            return jsonConverter;

        }
        @Bean
        public DefaultClassMapper classMapper() {
            DefaultClassMapper classMapper = new DefaultClassMapper();
            Map<String, Class<?>> idClassMapping = new HashMap<String, Class<?>>();
            idClassMapping.put("com.comcast.FileProcess.Pojo.CSVPojo", RequestPojo.class);
           // idClassMapping.put("bar", Bar.class);
            classMapper.setIdClassMapping(idClassMapping);
            return classMapper;
        }

        @Bean
        public Queue replyQueueRPC() {
            return new Queue("BulkSolve_GeneralreplyQueue");
        }

       @Bean
        public Queue requestQueueRPC() {
            return new Queue("BulkSolve_GeneralrequestQueue");
        }

        /*below are the newly added method for two other queues*/
        @Bean
        public Queue summaryreplyQueueRPC() {
            return new Queue("BulkSolve_SummaryreplyQueue");
        }

        @Bean
        public Queue summaryrequestQueueRPC() {
            return new Queue("BulkSolve_SummaryrequestQueue");
        }

        @Bean
        public SimpleMessageListenerContainer rpcGeneralReplyMessageListenerContainer(ConnectionFactory connectionFactory) {
            SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
            simpleMessageListenerContainer.setQueues(replyQueueRPC());
            simpleMessageListenerContainer.setTaskExecutor(taskExecutor());
            //simpleMessageListenerContainer.setMessageListener(listenerAdapter1);
            return simpleMessageListenerContainer;
        }
        @Bean
        public SimpleMessageListenerContainer rpcSummaryReplyMessageListenerContainer(ConnectionFactory connectionFactory) {
            SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
            simpleMessageListenerContainer.setQueues(summaryreplyQueueRPC());
            //simpleMessageListenerContainer.setMessageListener(listenerAdapter2);
            simpleMessageListenerContainer.setTaskExecutor(taskExecutor());
            return simpleMessageListenerContainer;
        }

       /* @Bean
        @Qualifier("listenerAdapter1")
        MessageListenerAdapter listenerAdapter1(SubscriberGeneralQueue1 generalReceiver) {
            return new MessageListenerAdapter(generalReceiver, "receivegeneralQueueMessage");
        }*/

      /*  @Bean
        @Qualifier("listenerAdapter2")
        MessageListenerAdapter listenerAdapter2(SubscriberSummaryQueue1 summaryReceiver) {
            return new MessageListenerAdapter(summaryReceiver, "receivesummaryQueueMessage");
        }*/


       @Bean
        public RequestPojo requestPojo(){
        return new RequestPojo();
        }
       /* @Bean
        @Qualifier("asyncGeneralRabbitTemplate")
        public AsyncRabbitTemplate asyncGeneralRabbitTemplate(ConnectionFactory connectionFactory) {

            AsyncRabbitTemplate asyncGeneralRabbitTemplate = new AsyncRabbitTemplate(rabbitTemplate(connectionFactory),
                            rpcGeneralReplyMessageListenerContainer(connectionFactory),
                            "spring-boot-rabbitmq-BulkSolve.async_BulkSolve_Msg" + "/" + "BulkSolve_GeneralreplyQueue");

            AsyncRabbitTemplate at = new AsyncRabbitTemplate(connectionFactory, "spring-boot-rabbitmq-examples.async_rpc", "rpc_request", "replyQueueRPC","replyQueueRPC");
            return asyncGeneralRabbitTemplate;
        }

        template defined for other 2 queues
        @Bean
        @Qualifier("asyncSummaryRabbitTemplate")
        public AsyncRabbitTemplate asyncSummaryRabbitTemplate(ConnectionFactory connectionFactory) {

            AsyncRabbitTemplate asyncSummaryRabbitTemplate = new AsyncRabbitTemplate(rabbitTemplate(connectionFactory),
                    rpcSummaryReplyMessageListenerContainer(connectionFactory),
                            "spring-boot-rabbitmq-BulkSolve_summary.async_BulkSolve_Msg" + "/" + "BulkSolve_SummaryreplyQueue");

            AsyncRabbitTemplate at = new AsyncRabbitTemplate(connectionFactory, "spring-boot-rabbitmq-examples.async_rpc", "rpc_request", "replyQueueRPC","replyQueueRPC");
            return asyncSummaryRabbitTemplate;
        }*/

        @Bean
        public DirectExchange directExchange() {
            return new DirectExchange("spring-boot-rabbitmq-BulkSolve.async_BulkSolve_Msg");
        }
      //Added new exchange
        @Bean
        public DirectExchange directExchange1() {
            return new DirectExchange("spring-boot-rabbitmq-BulkSolve_summary.async_BulkSolve_Msg");
        }
        @Bean
        public List<Binding> bindings() {
            return Arrays.asList(
                            BindingBuilder.bind(requestQueueRPC()).to(directExchange()).with("BulkSolve_GeneralrequestQueue"),
                            BindingBuilder.bind(replyQueueRPC()).to(directExchange()).with("BulkSolve_GeneralreplyQueue"),
                            BindingBuilder.bind(summaryrequestQueueRPC()).to(directExchange1()).with("BulkSolve_SummaryrequestQueue"),
                            BindingBuilder.bind(summaryreplyQueueRPC()).to(directExchange1()).with("BulkSolve_SummaryreplyQueue")
                            );


        }

}

//下面是我的读者课程

@Component
public class Reader extends AmqpItemReader<List<RequestPojo>>{

    @Autowired
     @Qualifier("rabbitTemplate")
     private RabbitTemplate rabbitTemplate;

    public static boolean batchstatus;

    private List<RequestPojo> reqList = new ArrayList<RequestPojo>();

/*  @Autowired
    private SimpleMessageListenerContainer messagelistener;*/

    public Reader(AmqpTemplate rabbitTemplate) {
        super(rabbitTemplate);
        // TODO Auto-generated constructor stub
    }

    List<RequestPojo> msgList = new ArrayList<RequestPojo>();
    @Override
    @SuppressWarnings("unchecked")
    public List<RequestPojo> read() {



        if(!batchstatus){
        RequestPojo msg=(RequestPojo)rabbitTemplate.receiveAndConvert("BulkSolve_GeneralrequestQueue" );
        //rabbitTemplate.receiveandco
            //return (List<RequestPojo>) rabbitTemplate.receive();

        System.out.println("I am inside Reader" + msg);
        msgList.add(msg);

        //Object result = rabbitTemplate.receiveAndConvert();

        batchstatus=true;

        return  msgList;
        }
        return null;
    }
}

下面是我的触发点代码,当队列中有消息时触发Job。

@EnableRabbit
public class Eventscheduler {

    @Autowired
    Job csvJob;

    @Autowired
    private JobLauncher jobLauncher;

    //@Scheduled(cron="0 */2 * ? * *")
    @RabbitListener(queues ="BulkSolve_GeneralrequestQueue")
    public void trigger(){
        Reader.batchstatus=false;
        Map<String,JobParameter> maps= new HashMap<String,JobParameter>();

        maps.put("time", new JobParameter(System.currentTimeMillis()));
        JobParameters jobParameters = new JobParameters(maps);
        JobExecution execution=null;
        try {
            //JobLauncher jobLauncher = new JobLauncher();
             execution=jobLauncher.run(csvJob, jobParameters);
        } catch (JobExecutionAlreadyRunningException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (JobRestartException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (JobInstanceAlreadyCompleteException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (JobParametersInvalidException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        System.out.println("JOB Executed:" + execution.getStatus());

    }

}

谢谢。

spring-boot spring-batch spring-amqp
1个回答
-1
投票

正如你注意到subscribeToRequestQueue(@Payload RequestPojo sampleRequestMessage)rabbitTemplate.receiveAndConvert("BulkSolve_GeneralrequestQueue")之间存在细微差别。而且,那个正是@Payload RequestPojo在原始的receiveAndConvert()错过了。因此,当执行此方法时,没有目标类型可以参考预期的转换。这样我们就可以回退到传入消息中的所有内容。在你的情况下,这是一个__TypeId__标头与生产者com.comcast.FileProcess.Pojo.CSVPojo的源类型。

如果你真的想在消费者方面强制转换为RequestPojo,你需要考虑使用重载的receiveAndConvert变体:

/**
 * Receive a message if there is one from a specific queue and convert it to a Java
 * object. Returns immediately, possibly with a null value. Requires a
 * {@link org.springframework.amqp.support.converter.SmartMessageConverter}.
 *
 * @param queueName the name of the queue to poll
 * @param type the type to convert to.
 * @param <T> the type.
 * @return a message or null if there is none waiting
 * @throws AmqpException if there is a problem
 * @since 2.0
 */
<T> T receiveAndConvert(String queueName, ParameterizedTypeReference<T> type) throws AmqpException;
© www.soinside.com 2019 - 2024. All rights reserved.