Spring AMQP 2.0：高级消息处理与依赖注入支持

需积分: 10 30 浏览量更新于2024-07-18 收藏 898KB PDF 举报

Spring AMQP 是一个基于Spring框架的高级库，它将核心Spring概念应用于AMQP（Advanced Message Queuing Protocol）消息解决方案的开发。这个库的主要目标是提供一种模板，作为发送和接收消息的高层抽象，以简化开发者在构建企业级消息驱动应用程序时的工作。它支持Message-driven POJOs（Plain Old Java Objects，即普通的Java对象）模式，这使得开发者可以利用依赖注入和声明式配置来管理AMQP资源，提高代码的灵活性和可维护性。 SpringAMQP 2.0.5.RELEASE 版本包含了多个重要功能和改进，由Mark Pollack、Mark Fisher等多位贡献者共同开发，版权归属Pivotal Software Inc.。以下是该版本的一些关键特性： 1. **兼容性**：Spring AMQP 2.0旨在与多个版本的Spring框架兼容，包括XML配置和Java配置两种方式，以便开发者根据项目需求选择最合适的配置方式。 2. **快速入门**：对于急于尝试的开发者，提供了快速上手指南，让新用户能快速了解如何在Spring环境中设置和使用AMQP。 3. **新功能**： - **CachingConnectionFactory**：引入了缓存连接工厂，可以提高连接池的效率，减少创建连接的开销。 - **AMQP客户端库**：增强了客户端的功能，可能包括新的API或优化的性能特性。 - **通用更改**：包括删除过时或不再推荐使用的类，以及对现有类的改进。 - **新ListenerContainer**：可能引入了一个新的监听器容器，用于更高效地管理消息监听器。 - **日志适配器**：支持Log4j和Logback等不同日志框架的集成，便于统一日志处理。 - **RabbitTemplate**：RabbitTemplate类可能有了更新，提供更便捷的消息发送和接收接口。 - **ListenerAdapter**：可能增强了ListenerAdapter的行为，使其更加灵活和易于定制。 - **ListenerContainer**：容器的回滚策略可能有了改进，如支持条件回滚，提高异常处理能力。 - **ConnectionFactory**：连接工厂的配置选项可能有所扩展或优化。 - **重试机制**：可能引入了更智能的重试策略，处理网络问题或其他临时错误。 - **匿名队列命名**：允许使用匿名队列名称，简化某些场景下的消息路由配置。 - **@RabbitListener**：注解可能新增了功能或提供了更好的类型安全。 4. **移除支持**：Jackson 1.x的支持被移除，建议开发者升级到更新的JSON库版本。 Spring AMQP 2.0.5.RELEASE 是一个功能增强且易于集成的库，它通过提供高级API和更好的配置选项，帮助开发者更高效地构建基于AMQP的消息驱动应用，同时保持了Spring框架的依赖注入和声明式编程的优势。

Spring AMQP

2.0.5.RELEASE Spring AMQP 9

3. Reference

This part of the reference documentation details the various components that comprise Spring AMQP.

The main chapter covers the core classes to develop an AMQP application. This part also includes a

chapter about the sample applications.

3.1 Using Spring AMQP

In this chapter, we will explore the interfaces and classes that are the essential components for

developing applications with Spring AMQP.

AMQP Abstractions

Introduction

Spring AMQP consists of a handful of modules, each represented by a JAR in the distribution.

These modules are: spring-amqp, and spring-rabbit. The spring-amqp module contains the

org.springframework.amqp.core package. Within that package, you will find the classes that

represent the core AMQP "model". Our intention is to provide generic abstractions that do not rely on

any particular AMQP broker implementation or client library. End user code will be more portable across

vendor implementations as it can be developed against the abstraction layer only. These abstractions

are then used implemented by broker-specific modules, such as spring-rabbit. There is currently only a

RabbitMQ implementation; however the abstractions have been validated in .NET using Apache Qpid

in addition to RabbitMQ. Since AMQP operates at the protocol level in principle, the RabbitMQ client

can be used with any broker that supports the same protocol version, but we do not test any other

brokers at present.

The overview here assumes that you are already familiar with the basics of the AMQP specification. If

you are not, then have a look at the resources listed in Chapter 5, Other Resources

Message

The 0-9-1 AMQP specification does not define a Message class or interface. Instead, when performing

an operation such as basicPublish(), the content is passed as a byte-array argument and additional

properties are passed in as separate arguments. Spring AMQP defines a Message class as part of

a more general AMQP domain model representation. The purpose of the Message class is to simply

encapsulate the body and properties within a single instance so that the API can in turn be simpler. The

Message class definition is quite straightforward.

public class Message {

private final MessageProperties messageProperties;

private final byte[] body;

public Message(byte[] body, MessageProperties messageProperties) {

this.body = body;

this.messageProperties = messageProperties;

}

public byte[] getBody() {

return this.body;

}

public MessageProperties getMessageProperties() {

return this.messageProperties;

}

Spring AMQP

2.0.5.RELEASE Spring AMQP 10

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.0.5.RELEASE Spring AMQP 12

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.0.5.RELEASE Spring AMQP 13

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();

剩余139页未读，继续阅读

pangmkin

粉丝: 0
资源: 4

Spring AMQP 2.0：高级消息处理与依赖注入支持

Spring整合RabbitMQ

spring整合rabbit mq的jar包

spring rabbitmq rpc 测试代码

Spring RabbitMQ demo

Spring rabbitmq集成

spring rabbitmq amqp

Spring RabbitMQ Actuator监控工具.zip

Spring RabbitMQ死信机制原理实例详解

Spring RabbitMQ：Java实现消息收发实战教程

Spring RabbitMQ Actuator监控工具：状态分析与健康检查

最新资源