SpringCloud Stream+RabbitMQ消息分区

忘是亡心i 2022-04-18 06:39 269阅读 0赞

本篇记录SpringCloud Stream+RabbitMQ 消息分区功能的实现。

# 消息分区介绍 #

有一些场景需要满足, 同一个特征的数据被同一个实例消费, 比如同一个id的传感器监测数据必须被同一个实例统计计算分析, 否则可能无法获取全部的数据.

假如我想让相同的消息都被同一个微服务结点来处理，但是我有4个服务节点组成负载均衡，通过消费分组的概念仍不能满足我的要求，所以Spring Cloud Stream又为了此类场景引入消息分区的概念。当生产者将消息数据发送给多个消费者实例时，保证同一消息数据始终是由同一个消费者实例接收和处理。

本篇中的三个项目和消息分组的三个项目是一样的，分别为：StreamProvider是消息生产端，StreamConsumer0和StreamConsumer1是消息消费端。

# 1 父maven工程 #

## 1.1 工程结构如下： ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70]

## 1.2 pom.xml如下： ##

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
      <modelVersion>4.0.0</modelVersion>
    
      <groupId>com.study</groupId>
      <artifactId>cloud-ma</artifactId>
      <version>0.0.1-SNAPSHOT</version>
      <packaging>pom</packaging>
      <name>SpringCloudStudy</name>
      <description>SpringCloudStudy</description>
    
      
      <repositories>          
        <repository>            
            <id>nexus</id>            
            <url>http://xxx.xx.xxx.xxx:8081/repository/maven-public/</url>            
            <releases>
                <enabled>true</enabled>
            </releases>           
            <snapshots>
                <enabled>true</enabled>
            </snapshots>          
          </repository>               
       </repositories>
    
      <parent>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-parent</artifactId>
         <version>2.0.3.RELEASE</version>
         <relativePath/>
      </parent>
    
      <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
         <java.version>1.8</java.version>
         <spring-cloud.version>Finchley.RELEASE</spring-cloud.version>
      </properties>
    
      <dependencies>
    
        
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
                <exclusions>
                    <exclusion>
                        <groupId>org.springframework.boot</groupId>
                        <artifactId>spring-boot-starter-logging</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
    
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-log4j2</artifactId>
            </dependency>
            
    
            
            <dependency>
                <groupId>com.fasterxml.jackson.dataformat</groupId>
                <artifactId>jackson-dataformat-yaml</artifactId>
            </dependency>
            
    
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    
        
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <optional>true</optional>
            <scope>true</scope>
        </dependency>
    
        
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.51</version>
        </dependency>
        
    
      </dependencies>
    
      <dependencyManagement>
          <dependencies>
              <dependency>
                  <groupId>org.springframework.cloud</groupId>
                  <artifactId>spring-cloud-dependencies</artifactId>
                  <version>${spring-cloud.version}</version>
                  <type>pom</type>
                  <scope>import</scope>
              </dependency>
          </dependencies>
      </dependencyManagement>
    
      <build>
         <plugins>
             <plugin>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-maven-plugin</artifactId>
                 <configuration>
                        
                        <fork>true</fork>
                 </configuration>
             </plugin>
         </plugins>
      </build>
    
    
      <modules>
          <module>EurekaServer</module>
          <module>EurekaClientHi</module>
        <module>EurekaClientRibbonCustomer</module>
        <module>EurekaClientHi2</module>
        <module>EurekaClientFeignCustomer</module>
        <module>EurekaClientZuul</module>
        <module>config_server</module>
        <module>config-client</module>
        <module>config-server-svn</module>
        <module>config-client-svn</module>
        <module>StreamProvider</module>
        <module>stream-output</module>
        <module>stream-input</module>
        <module>StreamRabbitMQSelf</module>
        <module>StreamConsumer0</module>
        <module>StreamConsumer1</module>
      </modules>
    </project>

# 2 StreamProvider工程节点（消息生产端） #

## 2.1 工程结构 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 1]

## 2.2 POM.xml ##

<?xml version="1.0"?>
    <project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
      <modelVersion>4.0.0</modelVersion>
      <parent>
        <groupId>com.study</groupId>
        <artifactId>cloud-ma</artifactId>
        <version>0.0.1-SNAPSHOT</version>
      </parent>
    
      <artifactId>StreamProvider</artifactId>
      <packaging>jar</packaging>
      <name>StreamProvider</name>
    
      <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
      </properties>
      <dependencies>
      
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
      
        <dependency>
          <groupId>junit</groupId>
          <artifactId>junit</artifactId>
          <scope>test</scope>
        </dependency>
      </dependencies>
    </project>

## 2.3 application.yml ##

server:
      port: 8089
    spring:
      cloud:
        stream:
          binders: 
            defaultRabbit: 
              type: rabbit
              environment:                                      #配置rabbimq连接环境
                spring: 
                  rabbitmq:
                    host: xxx.xxx.xxx.xxx
                    username: mazhen
                    password: mazhen
                    virtual-host: / 
          bindings: 
            msgSender:                                                     #生产者绑定，这个是消息通道的名称
              destination: exchange-msgSender                              #exchange名称,交换模式默认是topic;把SpringCloud stream的消息输出通道绑定到RabbitMQ的exchange-msg交换器。
              content-type: application/json
              producer: 
                partition-count: 2                                         #指定参与消息分区的消费端节点数量为2个
                partition-key-expression: headers['partitionKey']          #payload.id#这个是分区表达式, 例如当表达式的值为1, 那么在订阅者的instance-index中为1的接收方, 将会执行该消息.
            msgSender2:                                                    #生产者绑定，这个是消息通道的名称
              destination: exchange-msgSender                              #exchange名称,交换模式默认是topic;把SpringCloud stream的消息输出通道绑定到RabbitMQ的exchange-msgSender交换器。
              content-type: application/json
              producer: 
                partition-count: 2                                         #指定参与消息分区的消费端节点数量为2个
                partition-key-expression: headers['partitionKey']          #payload.id#这个是分区表达式, 例如当表达式的值为1, 那么在订阅者的instance-index中为1的接收方, 将会执行该消息.

partition-key-expression通过该参数指定了分区键的表达式规则,分区key的值是基于partitionKeyExpression计算得出的,用于每个消息被发送至对应分区的输出channel。

该表达式作用于传递给MessageChannel的send方法的参数，该参数是实现 org.springframework.messaging.Message接口的类,GenericMessage类是Spring为我们提供的一个实现Message接口的类，我们封装的信息将会放在payload属性上。

如果partitionKeyExpression的值是payload，将会使用整个我们放在GenericMessage中的信息做分区数据。payload 是消息的实体类型，可以为自定义类型,比如 User,Role等等。

在application.yml这个配置文件中，我们可以看到partition-key-expression的值是headers\['partitionKey'\]，而headers\['partitionKey'\]这个是由MessageBuilder类的setHeader()方法完成赋值的，详见：2.5.2 。

## 2.4 自定义通道 ##

/** * */
    package com.stream.provider.rabbitMQ.channels;
    
    import org.springframework.cloud.stream.annotation.Output;
    import org.springframework.messaging.MessageChannel;
    
    /** * @author mazhen * */
    public interface SendOutputChannel { 
    	// 这里可以定义不同的通道
        String MSG_SENDER  = "msgSender"; // 通道名
        String MSG_SENDER2 = "msgSender2"; // 通道名
        
        @Output(SendOutputChannel.MSG_SENDER)
        MessageChannel msgSender();
        
        @Output(SendOutputChannel.MSG_SENDER2)
        MessageChannel msgSender2();
    
    }

## 2.5 消息生产类 ##

### 2.5.1 消息生产类—接口 ###

/** * */
    package com.stream.provider.rabbitMQ.service;
    
    /** * @author mazhen * */
    public interface SendMsg { 
    	public void timerMessageSource();
    	public void sendMsgStr(String str);
    }

### 2.5.2 消息生产类—实现类 ###

/** * */
    package com.stream.provider.rabbitMQ.service.impl;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.integration.support.MessageBuilder;
    import org.springframework.messaging.Message;
    import org.springframework.scheduling.annotation.Scheduled;
    
    import com.stream.provider.rabbitMQ.channels.SendOutputChannel;
    import com.stream.provider.rabbitMQ.service.SendMsg;
    
    /** * @author mazhen * setHeader("partitionKey", 0)对partitionKey赋值为0，那么在 * application.yml中headers['partitionKey']的值就是0， * 那么在订阅者的instance-index中为0的接收方, 将会执行该消息. */
    @EnableBinding(value={ SendOutputChannel.class})
    public class SendMsgImpl implements SendMsg { 
       
    	private static Logger logger = LoggerFactory.getLogger(SendMsgImpl.class);
    	
    	@Autowired
    	private SendOutputChannel sendOutputChannel;
    	
    	
    	@Override
    	/** * 第一种方法, 没有指定output的MessageChannel, 通过OutputInterface去拿具体的Channel * 设置partitionKey主要是为了分区用, 可以根据根据这个partitionKey来分区 */
        @Scheduled(initialDelay = 1000, fixedRate = 5000)
    	public void timerMessageSource() { 
    		Message<String> message = MessageBuilder.withPayload("From timerMessageSource").setHeader("partitionKey", 1).build();
    		sendOutputChannel.msgSender().send(message);
    		logger.info("发送消息："+message.toString());
    	}
    	
    	@Override
    	public void sendMsgStr(String str) { 
    		if (!sendOutputChannel.msgSender().send(MessageBuilder.withPayload(str).setHeader("partitionKey", 0).build())) { 
    			 logger.error("生产者消息发送失败：" + str);
    		}
    		logger.info("[sendMsgStr]生产者消息发送:"+str);
    	}
    
    }

## 2.6 TestController ##

/** * */
    package com.stream.provider.rabbitMQ.controller;
    
    import java.io.IOException;
    
    import javax.servlet.http.HttpServletRequest;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestMethod;
    import org.springframework.web.bind.annotation.RestController;
    
    import com.stream.provider.rabbitMQ.service.SendMsg;
    import com.stream.provider.utils.common.ParameterUtil;
    
    
    
    
    /** * @author mazhen * */
    @RestController
    public class TestController { 
        
    	/** * 引入日志，注意都是"org.slf4j"包下 */
    	private final static Logger logger = LoggerFactory.getLogger(TestController.class);
    	
    	@Autowired
    	private SendMsg  sendMsg;
    	
    	@RequestMapping(value = "recevieCdkeyFrom",method = RequestMethod.POST)
    	public String recevieCdkeyFrom(HttpServletRequest request){ 
    		
    		String jsonStr = null;
    		try { 
    			jsonStr = ParameterUtil.getParametersStr(request);
    			logger.info("从合作方接收到的参数----:"+jsonStr);
    			sendMsg.sendMsgStr(jsonStr);
    		} catch (IOException e) { 
    			logger.error("异常信息："+e);
    			e.printStackTrace();
    			return "IOException:"+e;
    		}
    		return jsonStr;
    	}
    	
    }

## 2.7 启动类 ##

package com.stream.provider;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    /** * Hello world! * */
    @SpringBootApplication
    public class StreamProviderApplication { 
        public static void main( String[] args ) { 
            SpringApplication.run(StreamProviderApplication.class, args);
        }
    }

# 3 StreamConsumer0工程(消费端) #

## 3.1 工程结构 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 2]

## 3.2 POM.xml ##

<?xml version="1.0"?>
    <project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
      <modelVersion>4.0.0</modelVersion>
      <parent>
        <groupId>com.study</groupId>
        <artifactId>cloud-ma</artifactId>
        <version>0.0.1-SNAPSHOT</version>
      </parent>
    
      <artifactId>StreamConsumer0</artifactId>
      <name>StreamConsumer0</name>
    
      <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
      </properties>
      <dependencies>
      
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
       
        
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
      
        <dependency>
          <groupId>junit</groupId>
          <artifactId>junit</artifactId>
          <scope>test</scope>
        </dependency>
      </dependencies>
    </project>

## 3.3 application.yml ##

server:
      port: 8090
    spring:
      cloud:
        stream:
          binders: 
            defaultRabbit: 
              type: rabbit
              environment:                                      #配置rabbimq连接环境
                spring: 
                  rabbitmq:
                    host: xxx.xxx.xxx.xxx
                    username: mazhen
                    password: mazhen
                    virtual-host: / 
          bindings: 
            input:                                              #生产者绑定，这个是消息通道的名称
              group: group-A                                    #该项目节点为消息组group-A的一个消费端         
              destination: exchange-msgSender                   #exchange名称,交换模式默认是topic;把SpringCloud stream的消息输入通道绑定到RabbitMQ的exchange-msgSender交换器。
              content-type: application/json
              consumer: 
                partitioned: true                               #true 表示启用消息分区功能
          instance-count: 2                                     #表示消息分区的消费端节点数量为2个
          instance-index: 0                                     #该参数设置消费端实例的索引号，索引号从0开始。这里设置该节点的索引号为0

## 3.4 消息消费类 ##

### 3.4.1 消息消费类—接口 ###

/** * */
    package com.stream.consumer0.rabbitMQ.service;
    
    /** * @author mazhen * */
    public interface ReceviceMsg { 
    	public void receive(String payload);
    }

### 3.4.2 消息消费类—实现类 ###

/** * */
    package com.stream.consumer0.rabbitMQ.service.impl;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.annotation.StreamListener;
    import org.springframework.cloud.stream.messaging.Sink;
    
    import com.stream.consumer0.rabbitMQ.service.ReceviceMsg;
    
    /** * @author mazhen * */
    @EnableBinding(value = { Sink.class})
    public class ReceviceMsgImpl implements ReceviceMsg { 
    
    	private static Logger logger = LoggerFactory.getLogger(ReceviceMsgImpl.class);
    	
    	@StreamListener(Sink.INPUT)
    	@Override
    	public void receive(String payload) { 
    		logger.info("接收消息："+payload);
    	}
    }

## 3.5 启动类 ##

package com.stream.consumer0;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    /** * 启动类 * */
    @SpringBootApplication
    public class StreamConsumer0Application { 
        public static void main( String[] args ) { 
            SpringApplication.run(StreamConsumer0Application.class, args);
        }
    }

# 4 StreamConsumer1工程（消费端） #

## 4.1 工程结构 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 3]

## 4.2 POM.xml ##

<?xml version="1.0"?>
    <project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
      <modelVersion>4.0.0</modelVersion>
      <parent>
        <groupId>com.study</groupId>
        <artifactId>cloud-ma</artifactId>
        <version>0.0.1-SNAPSHOT</version>
      </parent>
    
      <artifactId>StreamConsumer1</artifactId>
      <name>StreamConsumer1</name>
    
      <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
      </properties>
      <dependencies>
      
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
       
        
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
      
        <dependency>
          <groupId>junit</groupId>
          <artifactId>junit</artifactId>
          <scope>test</scope>
        </dependency>
      </dependencies>
    </project>

## 4.3 application.yml ##

server:
      port: 8091
    spring:
      cloud:
        stream:
          binders: 
            defaultRabbit: 
              type: rabbit
              environment:                                      #配置rabbimq连接环境
                spring: 
                  rabbitmq:
                    host: xxx.xxx.xxx.xxx
                    username: mazhen
                    password: mazhen
                    virtual-host: / 
          bindings: 
            input:                                              #生产者绑定，这个是消息通道的名称
              group: group-A                                    #该项目节点为消息组group-A的一个消费端         
              destination: exchange-msgSender                   #exchange名称,交换模式默认是topic;把SpringCloud stream的消息输入通道绑定到RabbitMQ的exchange-msgSender交换器。
              content-type: application/json
              consumer: 
                partitioned: true                               #true 表示启用消息分区功能
          instance-count: 2                                     #表示消息分区的消费端节点数量为2个
          instance-index: 1                                     #该参数设置消费端实例的索引号，索引号从0开始。这里设置该节点的索引号为1

## 4.4 消息消费类 ##

### 4.4.1 消息消费类–接口 ###

/** * */
    package com.stream.consumer1.rabbitMQ.service;
    
    /** * @author mazhen * */
    public interface ReceviceMsg { 
    	public void receive(String payload);
    }

### 4.4.2 消息消费类–实现类 ###

/** * */
    package com.stream.consumer1.rabbitMQ.service.impl;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.annotation.StreamListener;
    import org.springframework.cloud.stream.messaging.Sink;
    
    import com.stream.consumer1.rabbitMQ.service.ReceviceMsg;
    
    /** * @author mazhen * */
    @EnableBinding(value = { Sink.class})
    public class ReceviceMsgImpl implements ReceviceMsg { 
    
    	private static Logger logger = LoggerFactory.getLogger(ReceviceMsgImpl.class);
    	
    	@StreamListener(Sink.INPUT)
    	@Override
    	public void receive(String payload) { 
    		logger.info("接收消息："+payload);
    	}
    }

## 4.5 启动类 ##

package com.stream.consumer1;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    /** * Hello world! * */
    @SpringBootApplication
    public class StreamConsumer1Application { 
        public static void main( String[] args ) { 
            SpringApplication.run(StreamConsumer1Application.class, args);
        }
    }

# 5 测试 #

*  启动RabbitMQ
 *  依次启动节点 StreamConsumer0 、 StreamConsumer1和StreamProvider

## 5.1 exchange-msgSender 交换器 ##

从下图中可以看到，RabbitMQ 中已经创建了 exchange-msgSender 交换器：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 4]

## 5.2 exchange-msgSender.group-A 消息队列 ##

RabbitMQ 中也已经创建了exchange-msgSender.group-A-0和exchange-msgSender.group-A-1 两个消息队列：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 5]

## 5.3 postman向StreamProvider发送请求并实现消息的生产 ##

### 5.3.1 postman向StreamProvider发送请求![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 6] ###

### 5.3.2 生产消息 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 7]

## 5.4 消费节点接收到的消息 ##

setHeader(“partitionKey”, 0)时，StreamConsumer0节点接收到消息：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 8]  
setHeader(“partitionKey”, 1)时，StreamConsumer1节点接收到消息：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 9]  
到这里，我们就完成了指定特定实例来消费信息（即消费分区）的功能。

#### 欢迎关注我的微信公众号，不定期更新文章和大家一起学习共勉- ####

![在这里插入图片描述][20191212162244101.png]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70]: /images/20220418/f907a57271f34a8ea834409b258f750f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 1]: /images/20220418/ed7c1a604f91446d89a6998092158d86.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 2]: /images/20220418/480ea9492ba64ffbaf3ed6034f643bc2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 3]: /images/20220418/180192bce65944bb86cbf5442b335942.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 4]: /images/20220418/bd63e17df9f64e619d49b4051b1a662e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 5]: /images/20220418/43709a9fbece4889a0bbd3225b67e4e5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 6]: /images/20220418/34e3f78ea78c4366862b7b4dab94a34d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 7]: /images/20220418/210c4960ee574a42a7492ca3cc441f47.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 8]: /images/20220418/864939d3c7c9407fb16d7f46875dc6ce.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 9]: /images/20220418/78bb09f39cf44bd5ad6f4b3c694bbec6.png
[20191212162244101.png]: /images/20220418/c28d719638bf417482a320058d6c9112.png