ps:本文只是简单一个整合介绍，属于抛砖引玉，具体实现还需大家深入研究哈..
1.首先是生产者配置
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
|
<?
xml
version
=
"1.0"
encoding
=
"UTF-8"
?>
<
beans
xmlns
=
"http://www.springframework.org/schema/beans"
xmlns:xsi
=
"http://www.w3.org/2001/XMLSchema-instance"
xmlns:context
=
"http://www.springframework.org/schema/context"
xmlns:rabbit
=
"http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">
<!-- 连接服务配置 -->
<
rabbit:connection-factory
id
=
"connectionFactory"
host
=
"localhost"
username
=
"guest"
password
=
"guest"
port
=
"5672"
/>
<
rabbit:admin
connection-factory
=
"connectionFactory"
/>
<!-- queue 队列声明-->
<
rabbit:queue
id
=
"queue_one"
durable
=
"true"
auto-delete
=
"false"
exclusive
=
"false"
name
=
"queue_one"
/>
<!-- exchange queue binging key 绑定 -->
<
rabbit:direct-exchange
name
=
"my-mq-exchange"
durable
=
"true"
auto-delete
=
"false"
id
=
"my-mq-exchange"
>
<
rabbit:bindings
>
<
rabbit:binding
queue
=
"queue_one"
key
=
"queue_one_key"
/>
</
rabbit:bindings
>
</
rabbit:direct-exchange
>
<-- spring amqp默认的是jackson 的一个插件,目的将生产者生产的数据转换为json存入消息队列，由于fastjson的速度快于jackson,这里替换为fastjson的一个实现 -->
<
bean
id
=
"jsonMessageConverter"
class
=
"mq.convert.FastJsonMessageConverter"
></
bean
>
<-- spring template声明-->
<
rabbit:template
exchange
=
"my-mq-exchange"
id
=
"amqpTemplate"
connection-factory
=
"connectionFactory"
message-converter
=
"jsonMessageConverter"
/>
</
beans
>
|
2.fastjson messageconver插件实现
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
|
import
org.apache.commons.logging.Log;
import
org.apache.commons.logging.LogFactory;
import
org.springframework.amqp.core.Message;
import
org.springframework.amqp.core.MessageProperties;
import
org.springframework.amqp.support.converter.AbstractMessageConverter;
import
org.springframework.amqp.support.converter.MessageConversionException;
import
fe.json.FastJson;
public
class
FastJsonMessageConverter
extends
AbstractMessageConverter {
private
static
Log log = LogFactory.getLog(FastJsonMessageConverter.
class
);
public
static
final
String DEFAULT_CHARSET =
"UTF-8"
;
private
volatile
String defaultCharset = DEFAULT_CHARSET;
public
FastJsonMessageConverter() {
super
();
//init();
}
public
void
setDefaultCharset(String defaultCharset) {
this
.defaultCharset = (defaultCharset !=
null
) ? defaultCharset
: DEFAULT_CHARSET;
}
public
Object fromMessage(Message message)
throws
MessageConversionException {
return
null
;
}
public
<T> T fromMessage(Message message,T t) {
String json =
""
;
try
{
json =
new
String(message.getBody(),defaultCharset);
}
catch
(UnsupportedEncodingException e) {
e.printStackTrace();
}
return
(T) FastJson.fromJson(json, t.getClass());
}
protected
Message createMessage(Object objectToConvert,
MessageProperties messageProperties)
throws
MessageConversionException {
byte
[] bytes =
null
;
try
{
String jsonString = FastJson.toJson(objectToConvert);
bytes = jsonString.getBytes(
this
.defaultCharset);
}
catch
(UnsupportedEncodingException e) {
throw
new
MessageConversionException(
"Failed to convert Message content"
, e);
}
messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
messageProperties.setContentEncoding(
this
.defaultCharset);
if
(bytes !=
null
) {
messageProperties.setContentLength(bytes.length);
}
return
new
Message(bytes, messageProperties);
}
}
|
3.生产者端调用
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
import
java.util.List;
import
org.springframework.amqp.core.AmqpTemplate;
public
class
MyMqGatway {
@Autowired
private
AmqpTemplate amqpTemplate;
public
void
sendDataToCrQueue(Object obj) {
amqpTemplate.convertAndSend(
"queue_one_key"
, obj);
}
}
|
4.消费者端配置(与生产者端大同小异)
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
|
<?
xml
version
=
"1.0"
encoding
=
"UTF-8"
?>
<
beans
xmlns
=
"http://www.springframework.org/schema/beans"
xmlns:xsi
=
"http://www.w3.org/2001/XMLSchema-instance"
xmlns:context
=
"http://www.springframework.org/schema/context"
xmlns:rabbit
=
"http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">
<!-- 连接服务配置 -->
<
rabbit:connection-factory
id
=
"connectionFactory"
host
=
"localhost"
username
=
"guest"
password
=
"guest"
port
=
"5672"
/>
<
rabbit:admin
connection-factory
=
"connectionFactory"
/>
<!-- queue 队列声明-->
<
rabbit:queue
id
=
"queue_one"
durable
=
"true"
auto-delete
=
"false"
exclusive
=
"false"
name
=
"queue_one"
/>
<!-- exchange queue binging key 绑定 -->
<
rabbit:direct-exchange
name
=
"my-mq-exchange"
durable
=
"true"
auto-delete
=
"false"
id
=
"my-mq-exchange"
>
<
rabbit:bindings
>
<
rabbit:binding
queue
=
"queue_one"
key
=
"queue_one_key"
/>
</
rabbit:bindings
>
</
rabbit:direct-exchange
>
<!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象-->
<
rabbit:listener-container
connection-factory
=
"connectionFactory"
acknowledge
=
"auto"
task-executor
=
"taskExecutor"
>
<
rabbit:listener
queues
=
"queue_one"
ref
=
"queueOneLitener"
/>
</
rabbit:listener-container
>
</
beans
>
|
5.消费者端调用
|
1
2
3
4
5
6
7
8
9
|
import
org.springframework.amqp.core.Message;
import
org.springframework.amqp.core.MessageListener;
public
class
QueueOneLitener
implements
MessageListener{
@Override
public
void
onMessage(Message message) {
System.out.println(
" data :"
+ message.getBody());
}
}
|
6.由于消费端当队列有数据到达时，对应监听的对象就会被通知到，无法做到批量获取，批量入库，因此可以在消费端缓存一个临时队列，将mq取出来的数据存入本地队列，后台线程定时批量处理即可
转自：http://blog.csdn.net/l192168134/article/details/51210188
