跳到主要内容

消息传递

DeepSeek V3 中英对照 Messaging

Spring Cloud Contract 允许您验证使用消息传递作为通信手段的应用程序。本文档中展示的所有集成都与 Spring 配合使用，但您也可以创建自己的集成并使用它。

消息传递 DSL 顶层元素​

消息传递的 DSL 与专注于 HTTP 的 DSL 略有不同。以下部分将解释这些差异：

由方法触发的输出​

输出消息可以通过调用一个方法来触发（例如，当合约启动时以及当消息发送时，调用 Scheduler），如下例所示：

def dsl = Contract.make {
	// Human readable description
	description 'Some description'
	// Label by means of which the output message can be triggered
	label 'some_label'
	// input to the contract
	input {
		// the contract will be triggered by a method
		triggeredBy('bookReturnedTriggered()')
	}
	// output message of the contract
	outputMessage {
		// destination to which the output message will be sent
		sentTo('output')
		// the body of the output message
		body('''{ "bookName" : "foo" }''')
		// the headers of the output message
		headers {
			header('BOOK-NAME', 'foo')
		}
	}
}
# Human readable description
description: Some description
# Label by means of which the output message can be triggered
label: some_label
input:
  # the contract will be triggered by a method
  triggeredBy: bookReturnedTriggered()
# output message of the contract
outputMessage:
  # destination to which the output message will be sent
  sentTo: output
  # the body of the output message
  body:
    bookName: foo
  # the headers of the output message
  headers:
    BOOK-NAME: foo

在之前的示例中，如果调用了名为 bookReturnedTriggered 的方法，输出消息会被发送到 output。在消息发布者端，我们生成一个测试来调用该方法以触发消息。在消费者端，你可以使用 some_label 来触发消息。

消费者/生产者​

important

本节内容仅适用于 Groovy DSL。

在HTTP中，存在client/stub和server/test的概念。您也可以在消息传递中使用这些范式。此外，Spring Cloud Contract Verifier 还提供了consumer和producer方法（请注意，您可以使用$或value方法来提供consumer和producer部分）。

通用​

在 input 或 outputMessage 部分，你可以调用 assertThat 并传入一个 method 的名称（例如 assertThatMessageIsOnTheQueue()），该方法已在基类中定义或通过静态导入引入。Spring Cloud Contract 会在生成的测试中运行该方法。

集成​

您可以使用以下任一集成配置：

  • Apache Camel

  • Spring Integration

  • Spring Cloud Stream

  • Spring JMS

由于我们使用的是 Spring Boot，如果你已将其中一个库添加到类路径中，所有消息传递配置都会自动设置。

important

请记住在生成测试的基类上添加 @AutoConfigureMessageVerifier 注解。否则，Spring Cloud Contract 的消息传递部分将无法正常工作。

important

如果你想使用 Spring Cloud Stream，请记得添加 org.springframework.cloud:spring-cloud-stream 的测试依赖，如下所示：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
    <type>test-jar</type>
    <scope>test</scope>
    <classifier>test-binder</classifier>
</dependency>
testImplementation(group: 'org.springframework.cloud', name: 'spring-cloud-stream', classifier: 'test-binder')

手动集成测试​

测试使用的主要接口是 org.springframework.cloud.contract.verifier.messaging.MessageVerifierSender 和 org.springframework.cloud.contract.verifier.messaging.MessageVerifierReceiver。它们定义了如何发送和接收消息。

在测试中，你可以注入一个 ContractVerifierMessageExchange 来发送和接收遵循契约的消息。然后在你的测试中添加 @AutoConfigureMessageVerifier。以下示例展示了如何操作：

@RunWith(SpringTestRunner.class)
@SpringBootTest
@AutoConfigureMessageVerifier
public static class MessagingContractTests {

  @Autowired
  private MessageVerifier verifier;
  ...
}
备注

如果你的测试也需要存根，那么 @AutoConfigureStubRunner 包含了消息传递配置，因此你只需要这一个注解。

生产者端消息测试生成​

在DSL中包含input或outputMessage部分会导致在发布者端创建测试。默认情况下，会创建JUnit 4测试。不过，也可以选择创建JUnit 5、TestNG或Spock测试。

important

传递给 messageFrom 或 sentTo 的目的地对于不同的消息传递实现可能有不同的含义。对于 Stream 和 Integration，它首先被解析为通道的 destination。然后，如果没有这样的 destination，它将被解析为通道名称。对于 Camel，它是一个特定的组件（例如，jms）。

考虑以下合约：

def contractDsl = Contract.make {
	name "foo"
	label 'some_label'
	input {
		triggeredBy('bookReturnedTriggered()')
	}
	outputMessage {
		sentTo('activemq:output')
		body('''{ "bookName" : "foo" }''')
		headers {
			header('BOOK-NAME', 'foo')
			messagingContentType(applicationJson())
		}
	}
}
label: some_label
input:
  triggeredBy: bookReturnedTriggered
outputMessage:
  sentTo: activemq:output
  body:
    bookName: foo
  headers:
    BOOK-NAME: foo
    contentType: application/json

对于前面的示例，将创建以下测试：

import com.jayway.jsonpath.DocumentContext;
import com.jayway.jsonpath.JsonPath;
import org.junit.Test;
import org.junit.Rule;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierObjectMapper;
import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierMessage;
import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierMessaging;

import static org.springframework.cloud.contract.verifier.assertion.SpringCloudContractAssertions.assertThat;
import static org.springframework.cloud.contract.verifier.util.ContractVerifierUtil.*;
import static com.toomuchcoding.jsonassert.JsonAssertion.assertThatJson;
import static org.springframework.cloud.contract.verifier.messaging.util.ContractVerifierMessagingUtil.headers;
import static org.springframework.cloud.contract.verifier.util.ContractVerifierUtil.fileToBytes;

public class FooTest {
	@Autowired ContractVerifierMessaging contractVerifierMessaging;
	@Autowired ContractVerifierObjectMapper contractVerifierObjectMapper;

	@Test
	public void validate_foo() throws Exception {
		// when:
			bookReturnedTriggered();

		// then:
			ContractVerifierMessage response = contractVerifierMessaging.receive("activemq:output",
					contract(this, "foo.yml"));
			assertThat(response).isNotNull();

		// and:
			assertThat(response.getHeader("BOOK-NAME")).isNotNull();
			assertThat(response.getHeader("BOOK-NAME").toString()).isEqualTo("foo");
			assertThat(response.getHeader("contentType")).isNotNull();
			assertThat(response.getHeader("contentType").toString()).isEqualTo("application/json");

		// and:
			DocumentContext parsedJson = JsonPath.parse(contractVerifierObjectMapper.writeValueAsString(response.getPayload()));
			assertThatJson(parsedJson).field("['bookName']").isEqualTo("foo");
	}

}
import com.jayway.jsonpath.DocumentContext
import com.jayway.jsonpath.JsonPath
import spock.lang.Specification
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierObjectMapper
import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierMessage
import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierMessaging

import static org.springframework.cloud.contract.verifier.assertion.SpringCloudContractAssertions.assertThat
import static org.springframework.cloud.contract.verifier.util.ContractVerifierUtil.*
import static com.toomuchcoding.jsonassert.JsonAssertion.assertThatJson
import static org.springframework.cloud.contract.verifier.messaging.util.ContractVerifierMessagingUtil.headers
import static org.springframework.cloud.contract.verifier.util.ContractVerifierUtil.fileToBytes

class FooSpec extends Specification {
	@Autowired ContractVerifierMessaging contractVerifierMessaging
	@Autowired ContractVerifierObjectMapper contractVerifierObjectMapper

	def validate_foo() throws Exception {
		when:
			bookReturnedTriggered()

		then:
			ContractVerifierMessage response = contractVerifierMessaging.receive("activemq:output",
					contract(this, "foo.yml"))
			response != null

		and:
			response.getHeader("BOOK-NAME") != null
			response.getHeader("BOOK-NAME").toString() == 'foo'
			response.getHeader("contentType") != null
			response.getHeader("contentType").toString() == 'application/json'

		and:
			DocumentContext parsedJson = JsonPath.parse(contractVerifierObjectMapper.writeValueAsString(response.getPayload()))
			assertThatJson(parsedJson).field("['bookName']").isEqualTo("foo")
	}

}

消费者存根生成​

与HTTP部分不同，在消息传递中，我们需要将合约定义与存根一同发布到JAR包中。随后在消费者端进行解析，并创建相应的存根路由。

important

如果类路径上存在多个框架，Stub Runner 需要定义应使用哪一个。假设类路径上有 AMQP、Spring Cloud Stream 和 Spring Integration，而您希望使用 Spring AMQP。那么您需要设置 spring.cloud.contract.stubrunner.stream.enabled=false 和 spring.cloud.contract.stubrunner.integration.enabled=false。这样，唯一剩下的框架就是 Spring AMQP。

桩触发​

要触发消息，请使用 StubTrigger 接口，如下例所示：

import java.util.Collection;
import java.util.Map;

/**
 * Contract for triggering stub messages.
 *
 * @author Marcin Grzejszczak
 */
public interface StubTrigger {

	/**
	 * Triggers an event by a given label for a given {@code groupid:artifactid} notation.
	 * You can use only {@code artifactId} too.
	 *
	 * Feature related to messaging.
	 * @param ivyNotation ivy notation of a stub
	 * @param labelName name of the label to trigger
	 * @return true - if managed to run a trigger
	 */
	boolean trigger(String ivyNotation, String labelName);

	/**
	 * Triggers an event by a given label.
	 *
	 * Feature related to messaging.
	 * @param labelName name of the label to trigger
	 * @return true - if managed to run a trigger
	 */
	boolean trigger(String labelName);

	/**
	 * Triggers all possible events.
	 *
	 * Feature related to messaging.
	 * @return true - if managed to run a trigger
	 */
	boolean trigger();

	/**
	 * Feature related to messaging.
	 * @return a mapping of ivy notation of a dependency to all the labels it has.
	 */
	Map<String, Collection<String>> labels();

}

为了方便起见，StubFinder 接口扩展了 StubTrigger，因此在你的测试中只需要使用其中一个即可。

StubTrigger 为您提供了以下触发消息的选项：

通过标签触发​

以下示例展示了如何触发带有标签的消息：

stubFinder.trigger('return_book_1')

通过 Group 和 Artifact ID 触发​

以下示例展示了如何通过组ID和工件ID触发消息：

stubFinder.trigger('org.springframework.cloud.contract.verifier.stubs:streamService', 'return_book_1')

通过制品ID触发​

以下示例展示了如何通过制品 ID 触发消息：

stubFinder.trigger('streamService', 'return_book_1')

触发所有消息​

以下示例展示了如何触发所有消息：

stubFinder.trigger()

使用 Apache Camel 进行消费者端消息传递​

Spring Cloud Contract Stub Runner 的消息模块为您提供了一种与 Apache Camel 轻松集成的方式。对于提供的构件，它会自动下载存根并注册所需的路由。

将 Apache Camel 添加到项目中​

您可以在类路径中同时包含 Apache Camel 和 Spring Cloud Contract Stub Runner。请记得使用 @AutoConfigureStubRunner 注解您的测试类。

禁用功能​

如果你需要禁用此功能，请设置 spring.cloud.contract.stubrunner.camel.enabled=false 属性。

示例​

假设我们有一个已部署 camelService 应用程序存根的 Maven 仓库：

└── .m2
    └── repository
        └── io
            └── codearte
                └── accurest
                    └── stubs
                        └── camelService
                            ├── 0.0.1-SNAPSHOT
                            │   ├── camelService-0.0.1-SNAPSHOT.pom
                            │   ├── camelService-0.0.1-SNAPSHOT-stubs.jar
                            │   └── maven-metadata-local.xml
                            └── maven-metadata-local.xml

进一步，假设存根包含以下结构：

├── META-INF
│   └── MANIFEST.MF
└── repository
    ├── accurest
    │   └── bookReturned1.groovy
    └── mappings

现在考虑以下合约：

Contract.make {
	label 'return_book_1'
	input {
		triggeredBy('bookReturnedTriggered()')
	}
	outputMessage {
		sentTo('rabbitmq:output?queue=output')
		body('''{ "bookName" : "foo" }''')
		headers {
			header('BOOK-NAME', 'foo')
		}
	}
}

要触发来自 return_book_1 标签的消息，我们使用 StubTrigger 接口，如下所示：

stubFinder.trigger("return_book_1")

这将向合约输出消息中描述的目的地发送一条消息。

使用 Spring Integration 进行消费者端消息传递​

Spring Cloud Contract Stub Runner 的消息模块为您提供了一种与 Spring Integration 集成的简便方式。对于提供的构件，它会自动下载存根并注册所需的路由。

将 Runner 添加到项目​

你可以在类路径中同时包含 Spring Integration 和 Spring Cloud Contract Stub Runner。记得用 @AutoConfigureStubRunner 注解你的测试类。

禁用功能​

如果你需要禁用此功能，请设置 spring.cloud.contract.stubrunner.integration.enabled=false 属性。

示例​

假设您拥有以下 Maven 仓库，其中部署了 integrationService 应用程序的存根：

└── .m2
    └── repository
        └── io
            └── codearte
                └── accurest
                    └── stubs
                        └── integrationService
                            ├── 0.0.1-SNAPSHOT
                            │   ├── integrationService-0.0.1-SNAPSHOT.pom
                            │   ├── integrationService-0.0.1-SNAPSHOT-stubs.jar
                            │   └── maven-metadata-local.xml
                            └── maven-metadata-local.xml

进一步假设存根包含以下结构：

├── META-INF
│   └── MANIFEST.MF
└── repository
    ├── accurest
    │   └── bookReturned1.groovy
    └── mappings

考虑以下合约：

Contract.make {
	label 'return_book_1'
	input {
		triggeredBy('bookReturnedTriggered()')
	}
	outputMessage {
		sentTo('output')
		body('''{ "bookName" : "foo" }''')
		headers {
			header('BOOK-NAME', 'foo')
		}
	}
}

现在考虑以下 Spring Integration 路由：

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
			 xmlns:beans="http://www.springframework.org/schema/beans"
			 xmlns="http://www.springframework.org/schema/integration"
			 xsi:schemaLocation="http://www.springframework.org/schema/beans
			https://www.springframework.org/schema/beans/spring-beans.xsd
			http://www.springframework.org/schema/integration
			http://www.springframework.org/schema/integration/spring-integration.xsd">

	<!-- REQUIRED FOR TESTING -->
	<bridge input-channel="output"
			output-channel="outputTest"/>

	<channel id="outputTest">
		<queue/>
	</channel>

</beans:beans>

要触发来自 return_book_1 标签的消息，请使用 StubTrigger 接口，如下所示：

stubFinder.trigger('return_book_1')

这将向合约输出消息中描述的目的地发送一条消息。

使用 Spring Cloud Stream 进行消费者端消息传递​

Spring Cloud Contract Stub Runner 的消息模块为您提供了一种与 Spring Stream 集成的简便方式。对于提供的构件，它会自动下载存根并注册所需的路由。

警告

如果 Stub Runner 与 Stream 的 messageFrom 或 sentTo 字符串首先被解析为通道的 destination 且不存在这样的 destination，则该目标将被解析为通道名称。

important

如果你想使用 Spring Cloud Stream，请记得添加对 org.springframework.cloud:spring-cloud-stream 测试支持的依赖，如下所示：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-test-binder</artifactId>
    <scope>test</scope>
</dependency>
testImplementation('org.springframework.cloud:spring-cloud-stream-test-binder')

将 Runner 添加到项目​

你可以在类路径中同时包含 Spring Cloud Stream 和 Spring Cloud Contract Stub Runner。记得用 @AutoConfigureStubRunner 注解你的测试类。

禁用功能​

如果你需要禁用此功能，请设置 spring.cloud.contract.stubrunner.stream.enabled=false 属性。

示例​

假设您有以下 Maven 仓库，其中部署了 streamService 应用程序的存根：

└── .m2
    └── repository
        └── io
            └── codearte
                └── accurest
                    └── stubs
                        └── streamService
                            ├── 0.0.1-SNAPSHOT
                            │   ├── streamService-0.0.1-SNAPSHOT.pom
                            │   ├── streamService-0.0.1-SNAPSHOT-stubs.jar
                            │   └── maven-metadata-local.xml
                            └── maven-metadata-local.xml

进一步假设存根包含以下结构：

├── META-INF
│   └── MANIFEST.MF
└── repository
    ├── accurest
    │   └── bookReturned1.groovy
    └── mappings

考虑以下合约：

Contract.make {
	label 'return_book_1'
	input { triggeredBy('bookReturnedTriggered()') }
	outputMessage {
		sentTo('returnBook')
		body('''{ "bookName" : "foo" }''')
		headers { header('BOOK-NAME', 'foo') }
	}
}

现在考虑以下 Spring Cloud Stream 函数配置：

@ImportAutoConfiguration(TestChannelBinderConfiguration.class)
@Configuration(proxyBeanMethods = true)
@EnableAutoConfiguration
protected static class Config {

	@Bean
	Function<String, String> test1() {
		return (input) -> {
			println "Test 1 [${input}]"
			return input
		}
	}

}

现在考虑以下 Spring 配置：

spring.cloud.contract.stubrunner.repositoryRoot: classpath:m2repo/repository/
spring.cloud.contract.stubrunner.ids: org.springframework.cloud.contract.verifier.stubs:streamService:0.0.1-SNAPSHOT:stubs
spring.cloud.contract.stubrunner.stubs-mode: remote
spring:
  cloud:
    stream:
      bindings:
        test1-in-0:
          destination: returnBook
        test1-out-0:
          destination: outputToAssertBook
    function:
      definition: test1

server:
  port: 0

debug: true

要触发来自 return_book_1 标签的消息，请使用 StubTrigger 接口，如下所示：

stubFinder.trigger('return_book_1')

这将向合约输出消息中描述的目的地发送一条消息。

使用 Spring JMS 进行消费者端消息传递​

Spring Cloud Contract Stub Runner 的消息模块提供了一种与 Spring JMS 集成的简便方式。

该集成假设您已有一个正在运行的 JMS 代理实例。

将 Runner 添加到项目​

你需要在类路径中同时包含 Spring JMS 和 Spring Cloud Contract Stub Runner。记得用 @AutoConfigureStubRunner 注解你的测试类。

示例​

假设存根结构如下所示：

├── stubs
    └── bookReturned1.groovy

进一步假设以下测试配置：

spring.cloud.contract.stubrunner:
  repository-root: stubs:classpath:/stubs/
  ids: my:stubs
  stubs-mode: remote
spring:
  activemq:
    send-timeout: 1000
  jms:
    template:
      receive-timeout: 1000

现在考虑以下合约：

Contract.make {
	label 'return_book_1'
	input {
		triggeredBy('bookReturnedTriggered()')
	}
	outputMessage {
		sentTo('output')
		body('''{ "bookName" : "foo" }''')
		headers {
			header('BOOKNAME', 'foo')
		}
	}
}

要触发来自 return_book_1 标签的消息，我们使用 StubTrigger 接口，如下所示：

stubFinder.trigger('return_book_1')

这将向合约输出消息中描述的目的地发送一条消息。