org.springframework.amqp.core.Message.getMessageProperties()方法的使用及代码示例

x33g5p2x  于2022-01-25 转载在 其他  
字(13.2k)|赞(0)|评价(0)|浏览(185)

本文整理了Java中org.springframework.amqp.core.Message.getMessageProperties()方法的一些代码示例，展示了Message.getMessageProperties()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Message.getMessageProperties()方法的具体详情如下：
包路径：org.springframework.amqp.core.Message
类名称：Message
方法名：getMessageProperties

Message.getMessageProperties介绍

暂无

代码示例

代码示例来源：origin: macrozheng/mall

  1. @Override
  2.   public Message postProcessMessage(Message message) throws AmqpException {
  3.     //给消息设置延迟毫秒值
  4.     message.getMessageProperties().setExpiration(String.valueOf(delayTimes));
  5.     return message;
  6.   }
  7. });

代码示例来源：origin: jmdhappy/xxpay-master

  1. public Message postProcessMessage(Message message) throws AmqpException {
  2.     message.getMessageProperties().setDelay((int) delay);
  3.     return message;
  4.   }
  5. });

代码示例来源：origin: qiurunze123/miaosha

  1. @RabbitListener(queues=MQConfig.MIAOSHATEST)
  2.   public void receiveMiaoShaMessage(Message message, Channel channel) throws IOException {
  3.     log.info("接受到的消息为:{}",message);
  4.     String messRegister = new String(message.getBody(), "UTF-8");
  5.     channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
  6.     MiaoShaMessageVo msm  = RedisService.stringToBean(messRegister, MiaoShaMessageVo.class);
  7.     messageService.insertMs(msm);
  8.     }
  9. }

代码示例来源：origin: openzipkin/brave

  1. TraceContextOrSamplingFlags extractAndClearHeaders(Message message) {
  2.  MessageProperties messageProperties = message.getMessageProperties();
  3.  TraceContextOrSamplingFlags extracted = extractor.extract(messageProperties);
  4.  Map<String, Object> headers = messageProperties.getHeaders();
  5.  clearHeaders(headers);
  6.  return extracted;
  7. }

代码示例来源：origin: yu199195/myth

  1. /**
  2.  * Message container simple message listener container.
  3.  *
  4.  * @return the simple message listener container
  5.  */
  6. @Bean
  7. @ConditionalOnProperty(prefix = "spring.rabbitmq", name = "host")
  8. public SimpleMessageListenerContainer messageContainer() {
  9.   SimpleMessageListenerContainer container =
  10.       new SimpleMessageListenerContainer(connectionFactory);
  11.   container.setQueues(queue());
  12.   container.setExposeListenerChannel(true);
  13.   container.setMaxConcurrentConsumers(2);
  14.   container.setConcurrentConsumers(1);
  15.   //设置确认模式手工确认
  16.   container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
  17.   container.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
  18.     byte[] messageBody = message.getBody();
  19.     LOGGER.debug("motan 框架接收到的消息");
  20.     //确认消息成功消费
  21.     final Boolean success = mythMqReceiveService.processMessage(messageBody);
  22.     if (success) {
  23.       channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
  24.     }
  25.   });
  26.   return container;
  27. }

代码示例来源：origin: yu199195/myth

  1. @Bean
  2. @ConditionalOnProperty(prefix = "spring.rabbitmq", name = "host")
  3. public SimpleMessageListenerContainer messageContainer() {
  4.   SimpleMessageListenerContainer container =
  5.       new SimpleMessageListenerContainer(connectionFactory);
  6.   container.setQueues(queue());
  7.   container.setExposeListenerChannel(true);
  8.   container.setMaxConcurrentConsumers(1);
  9.   container.setConcurrentConsumers(1);
  10.   //设置确认模式手工确认
  11.   container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
  12.   container.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
  13.     byte[] messageBody = message.getBody();
  14.     //确认消息成功消费
  15.     final Boolean success = mythMqReceiveService.processMessage(messageBody);
  16.     if (success) {
  17.       channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
  18.     }
  19.   });
  20.   return container;
  21. }

代码示例来源：origin: yu199195/myth

  1. /**
  2.  * Message container simple message listener container.
  3.  *
  4.  * @return the simple message listener container
  5.  */
  6. @Bean
  7. @ConditionalOnProperty(prefix = "spring.rabbitmq", name = "host")
  8. public SimpleMessageListenerContainer messageContainer() {
  9.   SimpleMessageListenerContainer container =
  10.       new SimpleMessageListenerContainer(connectionFactory);
  11.   container.setQueues(queue());
  12.   container.setExposeListenerChannel(true);
  13.   container.setMaxConcurrentConsumers(3);
  14.   container.setConcurrentConsumers(1);
  15.   //设置确认模式手工确认
  16.   container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
  17.   container.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
  18.     byte[] messageBody = message.getBody();
  19.     LogUtil.debug(LOGGER,()->"springcloud  account服务  amqp接收消息");
  20.     //确认消息成功消费
  21.     final Boolean success = mythMqReceiveService.processMessage(messageBody);
  22.     if (success) {
  23.       channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
  24.     }
  25.   });
  26.   return container;
  27. }

代码示例来源：origin: openzipkin/brave

  1. /**
  2.  * MethodInterceptor for {@link SimpleMessageListenerContainer.ContainerDelegate#invokeListener(Channel,
  3.  * Message)}
  4.  */
  5. @Override public Object invoke(MethodInvocation methodInvocation) throws Throwable {
  6.  Message message = (Message) methodInvocation.getArguments()[1];
  7.  TraceContextOrSamplingFlags extracted = springRabbitTracing.extractAndClearHeaders(message);
  8.  // named for BlockingQueueConsumer.nextMessage, which we can't currently see
  9.  Span consumerSpan = tracer.nextSpan(extracted);
  10.  Span listenerSpan = tracer.newChild(consumerSpan.context());
  11.  if (!consumerSpan.isNoop()) {
  12.   setConsumerSpan(consumerSpan, message.getMessageProperties());
  13.   // incur timestamp overhead only once
  14.   long timestamp = tracing.clock(consumerSpan.context()).currentTimeMicroseconds();
  15.   consumerSpan.start(timestamp);
  16.   long consumerFinish = timestamp + 1L; // save a clock reading
  17.   consumerSpan.finish(consumerFinish);
  18.   // not using scoped span as we want to start with a pre-configured time
  19.   listenerSpan.name("on-message").start(consumerFinish);
  20.  }
  21.  try (SpanInScope ws = tracer.withSpanInScope(listenerSpan)) {
  22.   return methodInvocation.proceed();
  23.  } catch (Throwable t) {
  24.   listenerSpan.error(t);
  25.   throw t;
  26.  } finally {
  27.   listenerSpan.finish();
  28.  }
  29. }

代码示例来源：origin: vector4wang/spring-boot-quick

  1. @Override
  2. public Message postProcessMessage(Message message) throws AmqpException {
  3.   message.getMessageProperties()
  4.       .setExpiration(ttl.toString()); // 设置per-message的失效时间
  5.   return message;
  6. }

代码示例来源：origin: openzipkin/brave

  1. @Override public Message postProcessMessage(Message message) {
  2.   TraceContext maybeParent = currentTraceContext.get();
  3.   // Unlike message consumers, we try current span before trying extraction. This is the proper
  4.   // order because the span in scope should take precedence over a potentially stale header entry.
  5.   //
  6.   // NOTE: Brave instrumentation used properly does not result in stale header entries, as we
  7.   // always clear message headers after reading.
  8.   Span span;
  9.   if (maybeParent == null) {
  10.    span = tracer.nextSpan(springRabbitTracing.extractAndClearHeaders(message));
  11.   } else {
  12.    // If we have a span in scope assume headers were cleared before
  13.    span = tracer.newChild(maybeParent);
  14.   }
  16.   if (!span.isNoop()) {
  17.    span.kind(Span.Kind.PRODUCER).name("publish");
  18.    if (remoteServiceName != null) span.remoteServiceName(remoteServiceName);
  19.    // incur timestamp overhead only once
  20.    long timestamp = tracing.clock(span.context()).currentTimeMicroseconds();
  21.    span.start(timestamp).finish(timestamp);
  22.   }
  24.   injector.inject(span.context(), message.getMessageProperties());
  25.   return message;
  26.  }
  27. }

代码示例来源：origin: FlowCI/flow-platform

  1. public PriorityMessage(Message message) {
  2.   super(message.getBody(), message.getMessageProperties());
  3.   this.timestamp = System.nanoTime();
  4. }

代码示例来源：origin: spring-projects/spring-integration

  1. protected void addDelayProperty(Message<?> message, org.springframework.amqp.core.Message amqpMessage) {
  2.   if (this.delayGenerator != null) {
  3.     amqpMessage.getMessageProperties().setDelay(this.delayGenerator.processMessage(message));
  4.   }
  5. }

代码示例来源：origin: vector4wang/spring-boot-quick

  1. @Override
  2.   public Message postProcessMessage(Message message) throws AmqpException {
  3.     message.getMessageProperties().setExpiration(msg.getTtl() + "");
  4.     return message;
  5.   }
  6. });

代码示例来源：origin: spring-projects/spring-integration

  1. private org.springframework.messaging.Message<Object> createMessage(Message message, Channel channel) {
  2.   Object payload = AmqpInboundChannelAdapter.this.messageConverter.fromMessage(message);
  3.   Map<String, Object> headers = AmqpInboundChannelAdapter.this.headerMapper
  4.       .toHeadersFromRequest(message.getMessageProperties());
  5.   if (AmqpInboundChannelAdapter.this.messageListenerContainer.getAcknowledgeMode()
  6.       == AcknowledgeMode.MANUAL) {
  7.     headers.put(AmqpHeaders.DELIVERY_TAG, message.getMessageProperties().getDeliveryTag());
  8.     headers.put(AmqpHeaders.CHANNEL, channel);
  9.   }
  10.   if (AmqpInboundChannelAdapter.this.retryTemplate != null) {
  11.     headers.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, new AtomicInteger());
  12.   }
  13.   final org.springframework.messaging.Message<Object> messagingMessage = getMessageBuilderFactory()
  14.       .withPayload(payload)
  15.       .copyHeaders(headers)
  16.       .build();
  17.   return messagingMessage;
  18. }

代码示例来源：origin: spring-projects/spring-integration

  1. protected Message<Object> buildMessage(org.springframework.amqp.core.Message message, Object converted) {
  2.   AbstractIntegrationMessageBuilder<Object> messageBuilder =
  3.       this.messageBuilderFactory.withPayload(converted);
  4.   if (this.channel.isExtractPayload()) {
  5.     Map<String, Object> headers =
  6.         this.inboundHeaderMapper.toHeadersFromRequest(message.getMessageProperties());
  7.     messageBuilder.copyHeaders(headers);
  8.   }
  9.   return messageBuilder.build();
  10. }

代码示例来源：origin: spring-projects/spring-integration

  1. private void send(String exchangeName, String routingKey,
  2.     final Message<?> requestMessage, CorrelationData correlationData) {
  3.   if (this.amqpTemplate instanceof RabbitTemplate) {
  4.     MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
  5.     org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, converter,
  6.         getHeaderMapper(), getDefaultDeliveryMode(), isHeadersMappedLast());
  7.     addDelayProperty(requestMessage, amqpMessage);
  8.     ((RabbitTemplate) this.amqpTemplate).send(exchangeName, routingKey, amqpMessage, correlationData);
  9.   }
  10.   else {
  11.     this.amqpTemplate.convertAndSend(exchangeName, routingKey, requestMessage.getPayload(),
  12.         message -> {
  13.           getHeaderMapper().fromHeadersToRequest(requestMessage.getHeaders(),
  14.               message.getMessageProperties());
  15.           return message;
  16.         });
  17.   }
  18. }

代码示例来源：origin: spring-projects/spring-integration

  1. protected AbstractIntegrationMessageBuilder<?> buildReply(MessageConverter converter,
  2.     org.springframework.amqp.core.Message amqpReplyMessage) {
  3.   Object replyObject = converter.fromMessage(amqpReplyMessage);
  4.   AbstractIntegrationMessageBuilder<?> builder = (replyObject instanceof Message)
  5.       ? this.getMessageBuilderFactory().fromMessage((Message<?>) replyObject)
  6.       : this.getMessageBuilderFactory().withPayload(replyObject);
  7.   Map<String, ?> headers = getHeaderMapper().toHeadersFromReply(amqpReplyMessage.getMessageProperties());
  8.   builder.copyHeadersIfAbsent(headers);
  9.   return builder;
  10. }

代码示例来源：origin: spring-projects/spring-integration

  1. @Test
  2. public void testAsyncDelayExpression() {
  3.   ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
  4.   AsyncRabbitTemplate amqpTemplate = spy(new AsyncRabbitTemplate(new RabbitTemplate(connectionFactory),
  5.       new SimpleMessageListenerContainer(connectionFactory), "replyTo"));
  6.   amqpTemplate.setTaskScheduler(mock(TaskScheduler.class));
  7.   AsyncAmqpOutboundGateway gateway = new AsyncAmqpOutboundGateway(amqpTemplate);
  8.   willAnswer(
  9.       invocation -> amqpTemplate.new RabbitMessageFuture("foo", invocation.getArgument(2)))
  10.         .given(amqpTemplate).sendAndReceive(anyString(), anyString(), any(Message.class));
  11.   gateway.setExchangeName("foo");
  12.   gateway.setRoutingKey("bar");
  13.   gateway.setDelayExpressionString("42");
  14.   gateway.setBeanFactory(mock(BeanFactory.class));
  15.   gateway.setOutputChannel(new NullChannel());
  16.   gateway.afterPropertiesSet();
  17.   gateway.start();
  18.   ArgumentCaptor<Message> captor = ArgumentCaptor.forClass(Message.class);
  19.   gateway.handleMessage(new GenericMessage<>("foo"));
  20.   verify(amqpTemplate).sendAndReceive(eq("foo"), eq("bar"), captor.capture());
  21.   assertThat(captor.getValue().getMessageProperties().getDelay(), equalTo(42));
  22. }

代码示例来源：origin: spring-projects/spring-integration

  1. @Test
  2. public void testHeaderMapperWinsAdapter() {
  3.   ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
  4.   RabbitTemplate amqpTemplate = spy(new RabbitTemplate(connectionFactory));
  5.   AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(amqpTemplate);
  6.   endpoint.setHeadersMappedLast(true);
  7.   final AtomicReference<Message> amqpMessage =
  8.       new AtomicReference<Message>();
  9.   willAnswer(invocation -> {
  10.     amqpMessage.set(invocation.getArgument(2));
  11.     return null;
  12.   }).given(amqpTemplate).send(isNull(), isNull(), any(Message.class), isNull());
  13.   org.springframework.messaging.Message<?> message = MessageBuilder.withPayload("foo")
  14.       .setHeader(MessageHeaders.CONTENT_TYPE, "bar")
  15.       .build();
  16.   endpoint.handleMessage(message);
  17.   assertNotNull(amqpMessage.get());
  18.   assertEquals("bar", amqpMessage.get().getMessageProperties().getContentType());
  19. }

代码示例来源：origin: spring-projects/spring-integration

  1. protected Message<?> buildReturnedMessage(org.springframework.amqp.core.Message message,
  2.     int replyCode, String replyText, String exchange, String routingKey, MessageConverter converter) {
  3.   Object returnedObject = converter.fromMessage(message);
  4.   AbstractIntegrationMessageBuilder<?> builder = (returnedObject instanceof Message)
  5.       ? this.getMessageBuilderFactory().fromMessage((Message<?>) returnedObject)
  6.       : this.getMessageBuilderFactory().withPayload(returnedObject);
  7.   Map<String, ?> headers = getHeaderMapper().toHeadersFromReply(message.getMessageProperties());
  8.   if (this.errorMessageStrategy == null) {
  9.     builder.copyHeadersIfAbsent(headers)
  10.         .setHeader(AmqpHeaders.RETURN_REPLY_CODE, replyCode)
  11.         .setHeader(AmqpHeaders.RETURN_REPLY_TEXT, replyText)
  12.         .setHeader(AmqpHeaders.RETURN_EXCHANGE, exchange)
  13.         .setHeader(AmqpHeaders.RETURN_ROUTING_KEY, routingKey);
  14.   }
  15.   Message<?> returnedMessage = builder.build();
  16.   if (this.errorMessageStrategy != null) {
  17.     returnedMessage = this.errorMessageStrategy.buildErrorMessage(new ReturnedAmqpMessageException(
  18.         returnedMessage, message, replyCode, replyText, exchange, routingKey), null);
  19.   }
  20.   return returnedMessage;
  21. }

内容来源于网络，如有侵权，请联系作者删除！

Message

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com