SpringCloudStream原理和深入使用小結(jié)

更新時間：2024年06月20日 12:29:59 作者：7仔要加油

Spring?Cloud?Stream是一個用于構(gòu)建與共享消息傳遞系統(tǒng)連接的高度可擴展的事件驅(qū)動型微服務(wù)的框架,本文給大家介紹SpringCloudStream原理和深入使用,感興趣的朋友跟隨小編一起看看吧

簡單概述

Spring Cloud Stream是一個用于構(gòu)建與共享消息傳遞系統(tǒng)連接的高度可擴展的事件驅(qū)動型微服務(wù)的框架。

應(yīng)用程序通過inputs或outputs來與Spring Cloud Stream中binder對象交互，binder對象負責(zé)與消息中間件交互。也就是說：Spring Cloud Stream能夠屏蔽底層消息中間件【RabbitMQ，kafka等】的差異，降低切換成本，統(tǒng)一消息的編程模型。

Spring Message

Spring Message是Spring Framework的一個模塊，其作用就是統(tǒng)一消息的編程模型。

package org.springframework.messaging;
public interface Message<T> {
    T getPayload();
    MessageHeaders getHeaders();
}

消息通道 MessageChannel 用于接收消息，調(diào)用send方法可以將消息發(fā)送至該消息通道中：

@FunctionalInterface
public interface MessageChannel {
	long INDEFINITE_TIMEOUT = -1;
	default boolean send(Message<?> message) {
		return send(message, INDEFINITE_TIMEOUT);
	}
	boolean send(Message<?> message, long timeout);
}

消息通道里的消息由消息通道的子接口可訂閱的消息通道SubscribableChannel實現(xiàn)，被MessageHandler消息處理器所訂閱

public interface SubscribableChannel extends MessageChannel {
	boolean subscribe(MessageHandler handler);
	boolean unsubscribe(MessageHandler handler);
}

由MessageHandler真正地消費/處理消息

@FunctionalInterface
public interface MessageHandler {
    void handleMessage(Message<?> message) throws MessagingException;
}

Spring Integration

Spring Integration 提供了 Spring 編程模型的擴展用來支持企業(yè)集成模式(Enterprise Integration Patterns)，是對 Spring Messaging 的擴展。

它提出了不少新的概念，包括消息路由MessageRoute、消息分發(fā)MessageDispatcher、消息過濾Filter、消息轉(zhuǎn)換Transformer、消息聚合Aggregator、消息分割Splitter等等。同時還提供了MessageChannel和MessageHandler的實現(xiàn)，分別包括 DirectChannel、ExecutorChannel、PublishSubscribeChannel和MessageFilter、ServiceActivatingHandler、MethodInvokingSplitter 等內(nèi)容。

Spring-Cloud-Stream的架構(gòu)

快速入門

引入依賴

        <!--stream-->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>

增加配置文件

spring:
    cloud:
        stream:
            # 定義消息中間件
            binders:
              MyRabbit:
                  type: rabbit
                  environment:
                    spring:
                        rabbitmq:
                            host: localhost
                            port: 5672
                            username: root
                            password: root
                            vhost: /
            bindings:
            # 生產(chǎn)者中定義，定義發(fā)布對象
              myInput:
                destination: myStreamExchange
                group: myStreamGroup
                binder: MyRabbit
            # 消費者中定義，定義訂閱的對象
              myOutput-in-0:
                destination: myStreamExchange
                group: myStreamGroup
                binder: MyRabbit
        # 消費者中定義，定義輸出的函數(shù)
        function:
            definition: myOutput

生產(chǎn)者

@Resource
	private StreamBridge streamBridge;
	public void sendNormal() {
		streamBridge.send("myInput", "hello world");
	}

消費者

@Bean("myOutput")
	public Consumer<Message<String>> myOutput() {
		return (message) -> {
			MessageHeaders headers = message.getHeaders();
			System.out.println("myOutput head is : " + headers);
			String payload = message.getPayload();
			System.out.println("myOutput payload is : " + payload);
		};
	}

如何自定義Binder

添加spring-cloud-stream依賴
提供ProvisioningProvider的實現(xiàn)提供
MessageProducer的實現(xiàn)提供
MessageHandler的實現(xiàn)提供
Binder的實現(xiàn)創(chuàng)建Binder的配置
在META-INF/spring.binders中定義綁定器

添加spring-cloud-stream依賴

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
    <version>${spring.cloud.stream.version}</version>
</dependency>

提供ProvisioningProvider的實現(xiàn)

ProvisioningProvider負責(zé)提供消費者和生產(chǎn)者目的地，并需要將 application.yml 或 application.properties 文件中包含的邏輯目的地轉(zhuǎn)換為物理目的地引用。

public class FileProvisioningProvider implements ProvisioningProvider<
	ExtendedConsumerProperties<FileConsumerProperties>, ExtendedProducerProperties<FileProducerProperties>> {
	public FileProvisioningProvider() {
		super();
	}
	@Override
	public ProducerDestination provisionProducerDestination(String name, ExtendedProducerProperties<FileProducerProperties> properties) throws ProvisioningException {
		return new FileMessageDestination(name);
	}
	@Override
	public ConsumerDestination provisionConsumerDestination(String name, String group, ExtendedConsumerProperties<FileConsumerProperties> properties) throws ProvisioningException {
		return new FileMessageDestination(name);
	}
	private static class FileMessageDestination implements ProducerDestination, ConsumerDestination {
		private final String destination;
		private FileMessageDestination(final String destination) {
			this.destination = destination;
		}
		@Override
		public String getName() {
			return destination.trim();
		}
		@Override
		public String getNameForPartition(int partition) {
			throw new UnsupportedOperationException("Partitioning is not implemented for file messaging.");
		}
	}
}

提供MessageProducer的實現(xiàn)

MessageProducer負責(zé)使用事件并將其作為消息處理，發(fā)送給配置為使用此類事件的客戶端應(yīng)用程序。

super.onInit();
		executorService = Executors.newScheduledThreadPool(1);
	}
	@Override
	public void doStart() {
		executorService.scheduleWithFixedDelay(() -> {
			String payload = getPayload();
			if (payload != null) {
				Message<String> receivedMessage = MessageBuilder.withPayload(payload).build();
				sendMessage(receivedMessage);
			}
		}, 0, 50, TimeUnit.MILLISECONDS);
	}
	@Override
	protected void doStop() {
		executorService.shutdownNow();
	}
	private String getPayload() {
		try {
			List<String> allLines = Files.readAllLines(Paths.get(fileExtendedBindingProperties.getPath() + File.separator + destination.getName() + ".txt"));
			String currentPayload = allLines.get(allLines.size() - 1);
			if (!currentPayload.equals(previousPayload)) {
				previousPayload = currentPayload;
				return currentPayload;
			}
		} catch (IOException e) {
			FileUtil.touch(new File(fileExtendedBindingProperties.getPath() + File.separator + destination.getName() + ".txt"));
		}
		return null;
	}
}

提供MessageHandler的實現(xiàn)

MessageHandler提供產(chǎn)生事件所需的邏輯。

public class FileMessageHandler extends AbstractMessageHandler {
	FileExtendedBindingProperties fileExtendedBindingProperties;
	ProducerDestination destination;
	public FileMessageHandler(ProducerDestination destination, FileExtendedBindingProperties fileExtendedBindingProperties) {
		this.destination = destination;
		this.fileExtendedBindingProperties = fileExtendedBindingProperties;
	}
	@Override
	protected void handleMessageInternal(Message<?> message) {
		try {
			if (message.getPayload() instanceof byte[]) {
				Files.write(Paths.get(fileExtendedBindingProperties.getPath() + File.separator + destination.getName() + ".txt"), (byte[]) message.getPayload());
			} else {
				throw new RuntimeException("處理消息失敗");
			}
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
	}
}

提供Binder的實現(xiàn)

提供自己的Binder抽象實現(xiàn):

擴展AbstractMessageChannelBinder類
將自定義的 ProvisioningProvider 指定為 AbstractMessageChannelBinder 的通用參數(shù)
重寫createProducerMessageHandler和createConsumerEndpoint方法

public class FileMessageChannelBinder extends AbstractMessageChannelBinder
	<ExtendedConsumerProperties<FileConsumerProperties>, ExtendedProducerProperties<FileProducerProperties>, FileProvisioningProvider>
	implements ExtendedPropertiesBinder<MessageChannel, FileConsumerProperties, FileProducerProperties> {
	FileExtendedBindingProperties fileExtendedBindingProperties;
	public FileMessageChannelBinder(String[] headersToEmbed, FileProvisioningProvider provisioningProvider, FileExtendedBindingProperties fileExtendedBindingProperties) {
		super(headersToEmbed, provisioningProvider);
		this.fileExtendedBindingProperties = fileExtendedBindingProperties;
	}
	@Override
	protected MessageHandler createProducerMessageHandler(ProducerDestination destination, ExtendedProducerProperties<FileProducerProperties> producerProperties, MessageChannel errorChannel) throws Exception {
		FileMessageHandler fileMessageHandler = new FileMessageHandler(destination, fileExtendedBindingProperties);
		return fileMessageHandler;
	}
	@Override
	protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group, ExtendedConsumerProperties<FileConsumerProperties> properties) throws Exception {
		FileMessageProducerAdapter fileMessageProducerAdapter = new FileMessageProducerAdapter(destination, fileExtendedBindingProperties);
		return fileMessageProducerAdapter;
	}
	@Override
	public FileConsumerProperties getExtendedConsumerProperties(String channelName) {
		return fileExtendedBindingProperties.getExtendedConsumerProperties(channelName);
	}
	@Override
	public FileProducerProperties getExtendedProducerProperties(String channelName) {
		return fileExtendedBindingProperties.getExtendedProducerProperties(channelName);
	}
	@Override
	public String getDefaultsPrefix() {
		return fileExtendedBindingProperties.getDefaultsPrefix();
	}
	@Override
	public Class<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass() {
		return fileExtendedBindingProperties.getExtendedPropertiesEntryClass();
	}
}

創(chuàng)建Binder的配置

嚴(yán)格要求創(chuàng)建一個 Spring 配置來初始化你的綁定器實現(xiàn)的 bean

@EnableConfigurationProperties(FileExtendedBindingProperties.class)
@Configuration
public class FileMessageBinderConfiguration {
	@Bean
	@ConditionalOnMissingBean
	public FileProvisioningProvider fileMessageBinderProvisioner() {
		return new FileProvisioningProvider();
	}
	@Bean
	@ConditionalOnMissingBean
	public FileMessageChannelBinder fileMessageBinder(FileProvisioningProvider fileMessageBinderProvisioner, FileExtendedBindingProperties fileExtendedBindingProperties) {
		return new FileMessageChannelBinder(null, fileMessageBinderProvisioner, fileExtendedBindingProperties);
	}
	@Bean
	public FileProducerProperties fileConsumerProperties() {
		return new FileProducerProperties();
	}
}

詳細的代碼見https://gitee.com/xiaovcloud/spring-cloud-stream

到此這篇關(guān)于SpringCloudStream原理和深入使用的文章就介紹到這了,更多相關(guān)SpringCloudStream原理內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: