Stream消息驱动

搭建SpringCloudStream发送消息接收消息案例

	
		org.springframework.cloud
		spring-cloud-starter-stream-rabbit
	
	
		org.springframework.boot
		spring-boot-starter-test
	

spring:
  cloud:
    stream:
      binders:
        defaultRabbit: # 定义binder
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: root
                password: manage
      bindings:
      
        output: # 发送方特有
          destination: e1 # 交换机名
          content-type: text/plain # 数据内容类型
          binder: defaultRabbit # 引用binder

        input: # 接收方特有
          destination: e1 # 交换机名
          content-type: text/plain # 数据内容类型
          binder: defaultRabbit # 引用binder
          group: g1 # 消费者分组，同组竞争，不同组重复
          # 持久化也靠分组

@SpringBootApplication
public class App {
	public static void main(String[] args) {
		SpringApplication.run(App.class);
	}
}

调用方法即可发送消息

@Service
@EnableBinding(Source.class)
public class SendService {
	@Resource
	private MessageChannel output;

	public void send(String s) {
		output.send(MessageBuilder.withPayload(s.getBytes()).build());
	}
}

放在容器中即可消费消息

@Service
@EnableBinding(Sink.class)
public class ReceiveService {
	@Resource
	private MessageChannel output;

	@StreamListener(Sink.INPUT)
	public void receive(Message message) {
		System.out.println(message);
	}
}

@SpringBootTest
@RunWith(SpringRunner.class)
public class AppTest {

	@Autowired
	SendService hs;

	@Test
	public void fun1() {
		hs.send("你好，沃德！");

		//休息一分钟以等待消费
		try {
			TimeUnit.MINUTES.sleep(1);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}

成功！