使用spring stream发送消息代码实例

为什么使用spring stream ?

spring stream 是用来做消息队列发送消息使用的。他隔离了各种消息队列的区别，使用统一的编程模型来发送消息。

目前支持：

  • rabbitmq
  • kafka
  • rocketmq

启动rocketmq

rocketmq 支持windows

start mqnamesrv.cmd
start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true

修改pom.xml

<dependency>
      <groupId>com.alibaba.cloud</groupId>
      <artifactId>spring-cloud-stream-binder-rocketmq</artifactId>
    </dependency>

增加发送接收JAVA代码

public interface InputOutput {
  String MAIL_OUTPUT = "mailOutput";
  String MAIL_INPUT = "mailInput";
  String OUTPUT = "output";
  String INPUT = "input";
  @Output(OUTPUT)
  MessageChannel output();
  @Input(INPUT)
  SubscribableChannel input();
  @Output(MAIL_OUTPUT)
  MessageChannel mailOutput();
  @Input(MAIL_INPUT)
  SubscribableChannel mailInput();
}

在应用上增加注解

@EnableBinding({InputOutput.class})

增加yml配置

spring:
  cloud:
    stream:
     rocketmq:
      binder:
       name-server: 127.0.0.1:9876
     bindings:
      output:
       destination: bpmmessage
       group: bpmmessage-group

      input:
       destination: bpmmessage
       group: bpmmessage-group-consumer

      mailOutput:
       destination: mail
       group: mail-group

      mailInput:
        destination: mail
        group: mail-group-consumer

编写代码收发消息：

MessageModel messageModel=new MessageModel();

    messageModel.setMsgType("mail");
    messageModel.setContent("helloworld");

    inputOutput.mailOutput().send( MessageBuilder.withPayload(
        "mail"
    ).build());

    inputOutput.output().send(
        MessageBuilder.withPayload(
            messageModel
        ).build()
    );

这里发送的是两类消息。

接收消息：

@Service
public class MessageListener {

  @StreamListener(InputOutput.INPUT)
  public void receive(MessageModel message) {
    System.err.println(message);
    System.err.println("ok");
  }

  @StreamListener(InputOutput.MAIL_INPUT)
  public void receive(String message) {
    System.err.println(message);
    System.err.println("ok");
  }
}

分别接收两类消息

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持我们。

赞 (0)

相关推荐

  • Spring Cloud Stream如何实现服务之间的通讯

    Spring Cloud Stream如何实现服务之间的通讯

    Spring Cloud Stream Srping cloud Bus的底层实现就是Spring Cloud Stream,Spring Cloud Stream的目的是用于构建基于消息驱动(或事件驱动)的微服务架构.Spring Cloud Stream本身对Spring Messaging.Spring Integration.Spring Boot Actuator.Spring Boot Externalized Configuration等模块进行封装(整合)和扩展,下面我们实现两个

  • Nginx根据url中的path动态转发到upstream的实现

    在Nginx中,有一些高级场景,需要根据url中的path参数,动态转发到不通的upstream 场景1 /svr1/xxxx?yyy 转发到 svr1:8080/xxxx?yyy /svr2/xxxx?yyy 转发到 svr2:8080/xxxx?yyy 配置如下: location ~* /(srv[1-9]+)/(.*)$ { allow all; proxy_pass http://$1/$2$is_args$args; proxy_set_header Host $host; prox

  • Spring Cloud Stream分区分组原理图解

    消息分组 通常在生产环境,我们的每个服务都不会以单节点的方式运行在生产环境,当同一个服务启动多个实例的时候,这些实例都会绑定到同一个消息通道的目标主题(Topic)上.默认情况下,当生产者发出一条消息到绑定通道上,这条消息会产生多个副本被每个消费者实例接收和处理,但是有些业务场景之下,我们希望生产者产生的消息只被其中一个实例消费,这个时候我们需要为这些消费者设置消费组来实现这样的功能. 当把消费者复制一份,发现2个都能收到消息 2个消费者都加入同一个消费者 发现只有一个能收到 消息分区 有一些场

  • Spring Cloud Stream异常处理过程解析

    应用处理 当消费者在处理接收到的消息时,有可能会由于某些原因而抛出异常.若希望对抛出来的异常进行处理的话,就需要采取一些异常处理手段,异常处理的方式可分为三种:应用层面的处理.系统层面的处理以及通过RetryTemplate进行处理. 本小节先来介绍较为常用的应用层面的异常处理方式,该方式又细分为局部处理和全局处理. 局部处理 Stream相关的配置内容如下: spring: cloud: stream: rocketmq: binder: name-server: 192.168.190.12

  • 如何在java 8 stream表达式实现if/else逻辑

    简介 在Stream处理中,我们通常会遇到if/else的判断情况,对于这样的问题我们怎么处理呢? 还记得我们在上一篇文章lambda最佳实践中提到,lambda表达式应该越简洁越好,不要在其中写臃肿的业务逻辑. 接下来我们看一个具体的例子. 传统写法 假如我们有一个1 to 10的list,我们想要分别挑选出奇数和偶数出来,传统的写法,我们会这样使用: public void inForEach(){ List<Integer> ints = Arrays.asList(1, 2, 3, 4

  • 详解Spring Cloud Stream使用延迟消息实现定时任务(RabbitMQ)

    我们在使用一些开源调度系统(比如:elastic-job等)的时候,对于任务的执行时间通常都是有规律性的,可能是每隔半小时执行一次,或者每天凌晨一点执行一次.然而实际业务中还存在另外一种定时任务,它可能需要一些触发条件才开始定时,比如:编写博文时候,设置2小时之后发送.对于这些开始时间不确定的定时任务,我们也可以通过Spring Cloud Stream来很好的处理. 为了实现开始时间不确定的定时任务触发,我们将引入延迟消息的使用.RabbitMQ中提供了关于延迟消息的插件,所以本文就来具体介绍

  • 如何在java 8 map中使用stream

    简介 Map是java中非常常用的一个集合类型,我们通常也需要去遍历Map去获取某些值,java 8引入了Stream的概念,那么我们怎么在Map中使用Stream呢? 基本概念 Map有key,value还有表示key,value整体的Entry. 创建一个Map: Map<String, String> someMap = new HashMap<>(); 获取Map的entrySet: Set<Map.Entry<String, String>> en

  • 使用spring stream发送消息代码实例

    为什么使用spring stream ? spring stream 是用来做消息队列发送消息使用的.他隔离了各种消息队列的区别,使用统一的编程模型来发送消息. 目前支持: rabbitmq kafka rocketmq 启动rocketmq rocketmq 支持windows start mqnamesrv.cmd start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true 修改pom.xml <dependency> &

  • Spring web集成rabbitmq代码实例

    这篇文章主要介绍了Spring web集成rabbitmq代码实例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 引入java包: 本项目中仅引入了四个java包:amqp-client-5.7.3.jar,spring-rabbit-2.2.2.RELEASE.jar,spring-retry-1.2.4.RELEASE.jar,spring-amqp-2.2.2.RELEASE.jar spring-rabbitmq.xml <?xml v

  • Spring 动态代理实现代码实例

    这篇文章主要介绍了Spring 动态代理实现代码实例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 基于jdk实现的动态代理 package com.proxy.daili; import com.proxy.daili.service.IModelMath; import java.lang.reflect.InvocationHandler; import java.lang.reflect.Method; import java.lang

  • spring boot 配置HTTPS代码实例

    这篇文章主要介绍了spring boot 配置HTTPS代码实例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 spring boot 版本是<version>1.5.8.RELEASE</version> 1.配置文件里,看下不要有空格=[不要有空格] 2.别名 ================ server.port=8095 server.ssl.key-store=*.pfx server.ssl.key-store-pa

  • Spring Bean装载方式代码实例解析

    Spring Bean装载方式代码实例解析

    这篇文章主要介绍了Spring Bean装载方式代码实例解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 Bean的装配方式 Bean的装配可以理解为依赖关系注入 基于XML的装配 a) 设值注入 i.要求: Bean 类必须提供一个默认的无参构造方法. Bean 类必须为需要注入的属性提供对应的setter方法. b) 构造注入 package com.itheima.assemble; import java.util.List; pub

  • SpringBoot整合WebSocket实现后端向前端发送消息的实例代码

    一.什么是 websocket 接口 使用 websocket 建立长连接,服务端和客户端可以互相通信,服务端只要有数据更新,就可以主动推给客户端. WebSocket 使得客户端和服务器之间的数据交换变得更加简单,允许服务端主动向客户端推送数据.在 WebSocket API 中,浏览器和服务器只需要完成一次握手,两者之间就直接可以创建持久性的连接,并进行双向数据传输.在 WebSocket API 中,浏览器和服务器只需要做一个握手的动作,然后,浏览器和服务器之间就形成了一条快速通道.两者之

  • Java HTTP协议收发MQ 消息代码实例详解

    1. 准备环境 在工程 POM 文件添加 HTTP Java 客户端的依赖. <dependency> <groupId>org.eclipse.jetty</groupId> <artifactId>jetty-client</artifactId> <version>9.3.4.RC1</version> </dependency> <dependency> <groupId>com

  • spring aop注解配置代码实例

    本文实例为大家分享了spring aop注解配置的具体代码,供大家参考,具体内容如下 Demo.java package cn.itcast.e_annotation; import javax.annotation.Resource; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.context.ApplicationContext; import org.springfra

  • Spring整合Quartz开发代码实例

    我们使用Spring整合Quartz开发,本实例采用数据库模式的demo. xml文件配置如下: <?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns

  • Spring cache整合redis代码实例

    Spring-Cache是Spring3.1引入的基于注解的缓存技术,本质上它并不是一个具体的缓存实现,而是一个对缓存使用的抽象,通过Spring AOP技术,在原有的代码上添加少量的注解来实现将这个方法转成缓存方法的效果. 本来想来个分析源码,奈何水平有限,先从实战搞起. 先引入依赖: <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-start

随机推荐