JarCastingmq-aliyun
基于阿里云MQ的封装
使用
生产者
<dependency>
<groupId>com.github.onepiecex</groupId>
<artifactId>mq-aliyun-producer-spring-boot-starter</artifactId>
<version>1.0.1</version>
</dependency>
消费者
<dependency>
<groupId>com.github.onepiecex</groupId>
<artifactId>mq-aliyun-consumer-spring-boot-starter</artifactId>
<version>1.0.1</version>
</dependency>
生产者
定义发送地址
@Pid("PID_TEST")
public enum TestProducer {
@To(topic = "TEST", tag = "dish.add")
DISH_ADD,
@To(topic = "TEST", tag = "dish.update")
DISH_UPDATE,
@To(topic = "TEST", tag = "dish.del")
DISH_DEL
}
有序的生产者
@Pid(value = "PID_MEICANYUN" ,ordered = true)
消息发送
@Autowired
private ProducerFactory producerFactory;
sendAsync 发送无序消息
orderSend 发送顺序消息
producerFactory.sendAsync(TestProducer.DISH_ADD, new Dish(1L, "name"));
producerFactory.sendAsync(TestProducer.DISH_UPDATE,
new Dish(2L, "name"),
new DeliveryOption("key").setDeliverTime(System.currentTimeMillis() + 1000 * 60));
producerFactory.sendAsync(TestProducer.DISH_DEL, 1L, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
//send success
//do some thing...
}
@Override
public void onException(OnExceptionContext context) {
//send fail
//do some thing...
}
});
发送顺序消息时 需要使用orderSend发送标记了ordered = true 的生产者
SendResult sendResult = producerFactory.orderSend(......);
消费者定义
使用声明式集中定义, 方便后期维护
public class Consumers implements ConsumerAble {
@Override
public void init(Ons ons) {
//订阅普通消息(无序)
ons.consumer("CID_TEST_DISH")
.subscribeTopic("TEST")
.subscribeTag("dish.add || dish.update",TestConsumer::dishAdd)
.subscribeTag("dish.del",TestConsumer::dishDel);
//订阅顺序消息
ons.consumerOrdered("CID_ORDER_TEST_DISH",new ConsumerOptional().setConsumeThread(10))
.subscribeTopic("ORDER_TEST")
.subscribeTag("dish.add || dish.update",TestConsumer::dishAdd)
.subscribeTag("dish.del",TestConsumer::dishDel);
}
}
//该类下面的所有消费者消费失败时重试5次
@Reconsume(5)
public class TestConsumer {
//重试3次
@Reconsume(3)
public Action dishAdd(Dish dish){
//do some thine
return Action.commit;
}
public Action dishDel(Long dishId, Message message){
//Message 为阿里云mq 原生消息体
//do some thing
return Action.commit;
}
}
重试次数
消费业务逻辑代码如果返回
Action.reconsume，或者 NULL，或者抛出异常，
消息都会走重试流程，默认重试 16 次，如果重试 16 次后，仍然失败，则消息丢弃。
第几次重试 每次重试间隔时间
1 10 秒
2 30 秒
3 1 分钟
4 2 分钟
5 3 分钟
6 4 分钟
7 5 分钟
8 6 分钟
9 7 分钟
10 8 分钟
11 9 分钟
12 10 分钟
13 20 分钟
14 30 分钟
15 1 小时
16 2 小时
最大重试次数小于等于16次，则重试时间间隔同上表描述。
最大重试次数大于16次，超过16次的重试时间间隔均为每次2小时。
配置文件 (application.yaml)
aliyun :
#阿里云授权KEY
accessKey : XXXXXXX
#阿里云秘钥
secretKey : XXXXXXXXXX
mq :
#topic、cid、pid的后缀 (用于区分 开发 生产模式)
suffix : _dev
producer :
#发送超时时间(毫秒) 缺省3000
sendTimeOut : 1000
#生产者路径,支持多个 以,分割
packages : com.mq.aliyun.example
consumer :
#扫描实现了ConsumerAble的类的路径,支持多个 以,分割
packages : com.mq.aliyun.example
#默认消费线程数 , 缺省20
defaultThread : 10
#集群模式: CLUSTERING ,广播模式: BROADCASTING, 缺省: CLUSTERING
defaultModel : CLUSTERING
#默认重试次数, 缺省16次, 最多16次
defaultMaxReconsume : 5
#顺序消息消费失败进行重试前的等待时间 单位(毫秒) , 缺省: 100
#仅顺序消息才会生效
suspendTime : 200
自定义序列化方式
默认使用 fastJson 进行序列化和反序列化
这里我拿 jackJson 序列化 和 反序列化举例子
自定义发送消息的序列化方式
- resources
- META-INF
- services
- com.github.mq.producer.ProducerSerialize
在 com.github.mq.producer.ProducerSerialize 里 指定你的实现类
public class MyProducerSerialize implements ProducerSerialize {
private static final ObjectMapper mapper = new ObjectMapper();
@Override
public byte[] objToBytes(Object object) {
//do some thine
try {
return mapper.writeValueAsBytes(object);
} catch (JsonProcessingException e) {
e.printStackTrace();
throw new RuntimeException("序列化失败");
}
}
}
自定义反序列化方式
首先定义一个PARAMETER注解
@WithArgumentExtractor(JackArgumentExtractor.class)
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.PARAMETER})
public @interface JackJson {
}
定义反序列化实现类
public class JackArgumentExtractor implements ArgumentExtractor {
private static final ObjectMapper mapper = new ObjectMapper();
@Override
public Result extract(Message message, Class parameterClass, Annotation annotation) {
byte[] body = message.getBody();
try {
return Results.next( mapper.readValue(body,parameterClass));
} catch (IOException e) {
e.printStackTrace();
}
return Results.end(Action.commit);
}
}
使用
public Action jackjson(@JackJson Dish dish){
//do some thine
return Action.commit;
}
也可以实现 DefaultArgumentExtractor, 替换默认的反序列化方式
- resources
- META-INF
- services
- com.github.mq.consumer.DefaultArgumentExtractor
public Action jackjson(Dish dish){
//do some thine
return Action.commit;
}
License
Copyright (C) 2017 onepiece.x, Inc.
This work is licensed under the Apache License, Version 2.0. See LICENSE for details.
