Spring AMQP 2.1.2 RELEASE：核心特性与更新概览

需积分: 0 2 浏览量更新于2024-07-10 收藏 971KB PDF 举报

"SpringAMQP参考指南2.1.2.RELEASE版，由MarkPollack, MarkFisher, OlegZhurakousky等多位专家编写，版权归属PivotalSoftwareInc.，允许非商业性的复制与分发。该文档涵盖了SpringAMQP的基本概念、新特性、配置方式以及各种组件的更新内容。" SpringAMQP是Spring框架的一个扩展，它提供了一套全面的工具集，用于简化Java应用程序与AdvancedMessageQueuingProtocol（AMQP）消息代理的交互。AMQP是一种开放标准，用于异步消息传递，具有高可扩展性和灵活性。 **1. 快速入门** 对于急于上手的开发者，SpringAMQP提供了快速入门指南。无论是使用XML配置还是Java配置，都可以轻松搭建基础架构。在XML配置中，开发者可以通过定义`<rabbit:connection-factory>`，`<rabbit:template>`等元素来配置连接工厂和模板。而在Java配置中，可以使用`RabbitTemplate`，`SimpleMessageListenerContainer`等类进行配置。 **2. 新特性** 2.1版本相对于2.0.4版本有以下更新： - AMQP客户端库：可能进行了性能优化或功能增强。 - 包结构变化：可能为了代码组织和模块化进行了调整。 - 发布者确认（PublisherConfirms）变化：增强了发送消息的可靠性，确保消息被正确接收。 - ListenerContainerFactory改进：提高了监听容器的灵活性和效率。 - BrokerEventListener：新增了对消息代理事件的监听能力。 - RabbitAdmin变化：可能增加了新的管理操作或属性。 - RabbitTemplate变化：可能扩展了模板类的功能，如添加了新的方法或优化了现有方法。 - 消息转换：可能引入了新的消息转换策略，以支持更多类型的数据交换。 - 管理REST API：可能改进了通过HTTP接口进行代理管理的体验。 - @RabbitListener变化：对注解驱动的消息监听器进行了增强，例如增加了异步返回处理。 - ConnectionFactoryBean变化：可能改进了连接工厂的配置和创建过程。 - ConnectionFactory变化：可能对连接工厂的底层实现进行了优化。 - ListenerContainer变化：可能提升了监听容器的性能和稳定性，例如增加了并发控制或错误处理。 **3. 兼容性** 文档中提到的兼容性问题通常涉及到SpringAMQP与不同版本的AMQP服务器、其他Spring框架组件以及Java运行环境的兼容性。开发者应确保所使用的SpringAMQP版本与目标环境相匹配，以避免潜在的兼容性问题。 SpringAMQP为开发人员提供了一个强大且灵活的框架，用于构建基于AMQP的消息驱动系统。其丰富的功能、易于使用的API和持续的更新，使得开发者能够更专注于业务逻辑，而不是底层消息传递的复杂性。

Spring AMQP

2.1.2.RELEASE Spring AMQP 8

The MessageProperties interface defines several common properties such as messageId,

timestamp, contentType, and several more. Those properties can also be extended with user-defined

headers by calling the setHeader(String key, Object value) method.

Important

Starting with versions 1.5.7, 1.6.11, 1.7.4, 2.0.0, if a message body is a

serialized Serializable java object, it is no longer deserialized (by default) when

performing toString() operations (such as in log messages). This is to prevent unsafe

deserialization. By default, only java.util and java.lang classes are deserialized. To

revert to the previous behavior, you can add allowable class/package patterns by invoking

Message.addWhiteListPatterns(...). A simple * wildcard is supported, for example

com.foo.*, *.MyClass. Bodies that cannot be deserialized will be represented by

byte[<size>] in log messages.

Exchange

The Exchange interface represents an AMQP Exchange, which is what a Message Producer sends to.

Each Exchange within a virtual host of a broker will have a unique name as well as a few other properties:

public interface Exchange {

String getName();

String getExchangeType();

boolean isDurable();

boolean isAutoDelete();

Map<String, Object> getArguments();

}

As you can see, an Exchange also has a type represented by constants defined in ExchangeTypes.

The basic types are: Direct, Topic, Fanout, and Headers. In the core package you will find

implementations of the Exchange interface for each of those types. The behavior varies across these

Exchange types in terms of how they handle bindings to Queues. For example, a Direct exchange

allows for a Queue to be bound by a fixed routing key (often the Queue’s name). A Topic exchange

supports bindings with routing patterns that may include the * and # wildcards for exactly-one and zero-

or-more, respectively. The Fanout exchange publishes to all Queues that are bound to it without taking

any routing key into consideration. For much more information about these and the other Exchange

types, check out Chapter 5, Other Resources.

Note

The AMQP specification also requires that any broker provide a "default" Direct Exchange that has

no name. All Queues that are declared will be bound to that default Exchange with their names

as routing keys. You will learn more about the default Exchange’s usage within Spring AMQP in

the section called “AmqpTemplate”.

Spring AMQP

2.1.2.RELEASE Spring AMQP 10

We also provide a BindingBuilder to facilitate a "fluent API" style.

Binding b = BindingBuilder.bind(someQueue).to(someTopicExchange).with("foo.*");

Note

The BindingBuilder class is shown above for clarity, but this style works well when using a static

import for the bind() method.

By itself, an instance of the Binding class is just holding the data about a connection. In other words, it is

not an "active" component. However, as you will see later in the section called “Configuring the broker”,

Binding instances can be used by the AmqpAdmin class to actually trigger the binding actions on the

broker. Also, as you will see in that same section, the Binding instances can be defined using Spring’s

@Bean-style within @Configuration classes. There is also a convenient base class which further

simplifies that approach for generating AMQP-related bean definitions and recognizes the Queues,

Exchanges, and Bindings so that they will all be declared on the AMQP broker upon application startup.

The AmqpTemplate is also defined within the core package. As one of the main components

involved in actual AMQP messaging, it is discussed in detail in its own section (see the section called

“AmqpTemplate”).

Connection and Resource Management

Introduction

Whereas the AMQP model we described in the previous section is generic and applicable to all

implementations, when we get into the management of resources, the details are specific to the broker

implementation. Therefore, in this section, we will be focusing on code that exists only within our "spring-

rabbit" module since at this point, RabbitMQ is the only supported implementation.

The central component for managing a connection to the RabbitMQ broker is the ConnectionFactory

interface. The responsibility of a ConnectionFactory implementation is to provide an

instance of org.springframework.amqp.rabbit.connection.Connection which is a wrapper

for com.rabbitmq.client.Connection. The only concrete implementation we provide is

CachingConnectionFactory which, by default, establishes a single connection proxy that can be

shared by the application. Sharing of the connection is possible since the "unit of work" for messaging

with AMQP is actually a "channel" (in some ways, this is similar to the relationship between a Connection

and a Session in JMS). As you can imagine, the connection instance provides a createChannel

method. The CachingConnectionFactory implementation supports caching of those channels, and

it maintains separate caches for channels based on whether they are transactional or not. When creating

an instance of CachingConnectionFactory, the hostname can be provided via the constructor. The

username and password properties should be provided as well. If you would like to configure the size

of the channel cache (the default is 25), you could call the setChannelCacheSize() method here

as well.

Starting with version 1.3, the CachingConnectionFactory can be configured to cache connections

as well as just channels. In this case, each call to createConnection() creates a new connection

(or retrieves an idle one from the cache). Closing a connection returns it to the cache (if the cache

size has not been reached). Channels created on such connections are cached too. The use of

separate connections might be useful in some environments, such as consuming from an HA cluster,

in conjunction with a load balancer, to connect to different cluster members. Set the cacheMode to

CacheMode.CONNECTION.

Spring AMQP

2.1.2.RELEASE Spring AMQP 11

Note

This does not limit the number of connections, it specifies how many idle open connections are

allowed.

Starting with version 1.5.5, a new property connectionLimit is provided. When this is

set, it limits the total number of connections allowed. When set, if the limit is reached, the

channelCheckoutTimeLimit is used to wait for a connection to become idle. If the time is exceeded,

an AmqpTimeoutException is thrown.

Important

When the cache mode is CONNECTION, automatic declaration of queues etc. (See the section

called “Automatic Declaration of Exchanges, Queues and Bindings”) is NOT supported.

Also, at the time of writing, the rabbitmq-client library creates a fixed thread pool for each

connection (5 threads) by default. When using a large number of connections, you should consider

setting a custom executor on the CachingConnectionFactory. Then, the same executor

will be used by all connections and its threads can be shared. The executor’s thread pool should

be unbounded, or set appropriately for the expected utilization (usually, at least one thread per

connection). If multiple channels are created on each connection then the pool size will affect the

concurrency, so a variable (or simple cached) thread pool executor would be most suitable.

It is important to understand that the cache size is (by default) not a limit, but merely the number of

channels that can be cached. With a cache size of, say, 10, any number of channels can actually be in

use. If more than 10 channels are being used and they are all returned to the cache, 10 will go in the

cache; the remainder will be physically closed.

Starting with version 1.6, the default channel cache size has been increased from 1 to 25. In high volume,

multi-threaded, environments, a small cache means that channels are created and closed at a high rate.

Increasing the default cache size will avoid this overhead. You should monitor the channels in use via

the RabbitMQ Admin UI and consider increasing the cache size further if you see many channels being

created and closed. The cache will only grow on-demand (to suit the concurrency requirements of the

application) so this change will not impact existing low-volume applications.

Starting with version 1.4.2, the CachingConnectionFactory has a property

channelCheckoutTimeout. When this property is greater than zero, the channelCacheSize

becomes a limit on the number of channels that can be created on a connection. If the limit is reached,

calling threads will block until a channel is available or this timeout is reached, in which case a

AmqpTimeoutException is thrown.

Warning

Channels used within the framework (e.g. RabbitTemplate) will be reliably returned to the

cache. If you create channels outside of the framework, (e.g. by accessing the connection(s)

directly and invoking createChannel()), you must return them (by closing) reliably, perhaps in

a finally block, to avoid running out of channels.

CachingConnectionFactory connectionFactory = new CachingConnectionFactory("somehost");

connectionFactory.setUsername("guest");

connectionFactory.setPassword("guest");

Connection connection = connectionFactory.createConnection();

Spring AMQP

2.1.2.RELEASE Spring AMQP 12

When using XML, the configuration might look like this:

<bean id="connectionFactory"

class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">

<constructor-arg value="somehost"/>

</bean>

Note

There is also a SingleConnectionFactory implementation which is only available in the unit

test code of the framework. It is simpler than CachingConnectionFactory since it does not

cache channels, but it is not intended for practical usage outside of simple tests due to its lack of

performance and resilience. If you find a need to implement your own ConnectionFactory for

some reason, the AbstractConnectionFactory base class may provide a nice starting point.

A ConnectionFactory can be created quickly and conveniently using the rabbit namespace:

<rabbit:connection-factory id="connectionFactory"/>

In most cases this will be preferable since the framework can choose the best defaults for you. The

created instance will be a CachingConnectionFactory. Keep in mind that the default cache size for

channels is 25. If you want more channels to be cached set a larger value via the channelCacheSize

property. In XML it would look like this:

<bean id="connectionFactory"

class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">

<constructor-arg value="somehost"/>

</bean>

And with the namespace you can just add the channel-cache-size attribute:

<rabbit:connection-factory

id="connectionFactory" channel-cache-size="50"/>

The default cache mode is CHANNEL, but you can configure it to cache connections instead; in this

case, we use connection-cache-size:

<rabbit:connection-factory

id="connectionFactory" cache-mode="CONNECTION" connection-cache-size="25"/>

Host and port attributes can be provided using the namespace

<rabbit:connection-factory

id="connectionFactory" host="somehost" port="5672"/>

Alternatively, if running in a clustered environment, use the addresses attribute.

<rabbit:connection-factory

id="connectionFactory" addresses="host1:5672,host2:5672"/>

Here’s an example with a custom thread factory that prefixes thread names with rabbitmq-.

剩余149页未读，继续阅读

helloxzy

粉丝: 0
资源: 5

Spring AMQP 2.1.2 RELEASE：核心特性与更新概览

spring-amqp-2.0.3.RELEASE.jar

spring-amqp-1.4.0.RELEASE.jar

spring-amqp 相关jar

spring-amqp对应springboot版本

spring-boot-starter-parent 2.7.16-SNAPSHOT 对应 spring-boot-starter-amqp 版本选择

spring-amqp:2.2.19.re lease.jar 升级到 2.3.2

spring-boot-starter-amqp+application.yml的代码案例

spring-boot-starter-amqp 延时消息

spring-boot-starter-amqp+application.yml+controller的代码案例

最新资源