mirror of
https://gitee.com/mirrors/Spring-Cloud-Alibaba.git
synced 2021-06-26 13:25:11 +08:00
update rocketmq examples
This commit is contained in:
fangjian0423
@@ -0,0 +1,54 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-alibaba-examples</artifactId>
|
||||
<version>0.2.2.BUILD-SNAPSHOT</version>
|
||||
<relativePath>../../pom.xml</relativePath>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
|
||||
<artifactId>rocketmq-produce-example</artifactId>
|
||||
<packaging>jar</packaging>
|
||||
<description>Example demonstrating how to use rocketmq produce</description>
|
||||
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-web</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-actuator</artifactId>
|
||||
</dependency>
|
||||
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-deploy-plugin</artifactId>
|
||||
<version>${maven-deploy-plugin.version}</version>
|
||||
<configuration>
|
||||
<skip>true</skip>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
</project>
|
||||
@@ -0,0 +1,39 @@
|
||||
package org.springframework.cloud.alibaba.cloud.examples;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class Foo {
|
||||
|
||||
private int id;
|
||||
private String bar;
|
||||
|
||||
public Foo() {
|
||||
}
|
||||
|
||||
public Foo(int id, String bar) {
|
||||
this.id = id;
|
||||
this.bar = bar;
|
||||
}
|
||||
|
||||
public int getId() {
|
||||
return id;
|
||||
}
|
||||
|
||||
public void setId(int id) {
|
||||
this.id = id;
|
||||
}
|
||||
|
||||
public String getBar() {
|
||||
return bar;
|
||||
}
|
||||
|
||||
public void setBar(String bar) {
|
||||
this.bar = bar;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "Foo{" + "id=" + id + ", bar='" + bar + '\'' + '}';
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
package org.springframework.cloud.alibaba.cloud.examples;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.CommandLineRunner;
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.cloud.alibaba.cloud.examples.RocketMQProduceApplication.MySource;
|
||||
import org.springframework.cloud.stream.annotation.EnableBinding;
|
||||
import org.springframework.cloud.stream.annotation.Output;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
@SpringBootApplication
|
||||
@EnableBinding({ MySource.class })
|
||||
public class RocketMQProduceApplication {
|
||||
|
||||
public interface MySource {
|
||||
@Output("output1")
|
||||
MessageChannel output1();
|
||||
|
||||
@Output("output2")
|
||||
MessageChannel output2();
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(RocketMQProduceApplication.class, args);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public CustomRunner customRunner() {
|
||||
return new CustomRunner();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public CustomRunnerWithTransactional customRunnerWithTransactional() {
|
||||
return new CustomRunnerWithTransactional();
|
||||
}
|
||||
|
||||
public static class CustomRunner implements CommandLineRunner {
|
||||
@Autowired
|
||||
private SenderService senderService;
|
||||
|
||||
@Override
|
||||
public void run(String... args) throws Exception {
|
||||
int count = 5;
|
||||
for (int index = 1; index <= count; index++) {
|
||||
String msgContent = "msg-" + index;
|
||||
if (index % 3 == 0) {
|
||||
senderService.send(msgContent);
|
||||
}
|
||||
else if (index % 3 == 1) {
|
||||
senderService.sendWithTags(msgContent, "tagStr");
|
||||
}
|
||||
else {
|
||||
senderService.sendObject(new Foo(index, "foo"), "tagObj");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public static class CustomRunnerWithTransactional implements CommandLineRunner {
|
||||
@Autowired
|
||||
private SenderService senderService;
|
||||
|
||||
@Override
|
||||
public void run(String... args) throws Exception {
|
||||
// COMMIT_MESSAGE message
|
||||
senderService.sendTransactionalMsg("transactional-msg1", 1);
|
||||
// ROLLBACK_MESSAGE message
|
||||
senderService.sendTransactionalMsg("transactional-msg2", 2);
|
||||
// ROLLBACK_MESSAGE message
|
||||
senderService.sendTransactionalMsg("transactional-msg3", 3);
|
||||
// COMMIT_MESSAGE message
|
||||
senderService.sendTransactionalMsg("transactional-msg4", 4);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
package org.springframework.cloud.alibaba.cloud.examples;
|
||||
|
||||
import java.util.stream.Collectors;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
import org.apache.rocketmq.common.message.MessageConst;
|
||||
import org.apache.rocketmq.spring.support.RocketMQHeaders;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.cloud.alibaba.cloud.examples.RocketMQProduceApplication.MySource;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.util.MimeTypeUtils;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
@Service
|
||||
public class SenderService {
|
||||
|
||||
@Autowired
|
||||
private MySource source;
|
||||
|
||||
public void send(String msg) throws Exception {
|
||||
source.output1().send(MessageBuilder.withPayload(msg).build());
|
||||
}
|
||||
|
||||
public <T> void sendWithTags(T msg, String tag) throws Exception {
|
||||
Message message = MessageBuilder.createMessage(msg,
|
||||
new MessageHeaders(Stream.of(tag).collect(Collectors
|
||||
.toMap(str -> MessageConst.PROPERTY_TAGS, String::toString))));
|
||||
source.output1().send(message);
|
||||
}
|
||||
|
||||
public <T> void sendObject(T msg, String tag) throws Exception {
|
||||
Message message = MessageBuilder.withPayload(msg)
|
||||
.setHeader(MessageConst.PROPERTY_TAGS, tag)
|
||||
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
|
||||
.build();
|
||||
source.output1().send(message);
|
||||
}
|
||||
|
||||
public <T> void sendTransactionalMsg(T msg, int num) throws Exception {
|
||||
MessageBuilder builder = MessageBuilder.withPayload(msg)
|
||||
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON);
|
||||
builder.setHeader("test", String.valueOf(num));
|
||||
builder.setHeader(RocketMQHeaders.TAGS, "binder");
|
||||
Message message = builder.build();
|
||||
source.output2().send(message);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
/*
|
||||
* Copyright (C) 2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.cloud.alibaba.cloud.examples;
|
||||
|
||||
import java.util.HashMap;
|
||||
|
||||
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
|
||||
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
|
||||
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
|
||||
import org.apache.rocketmq.spring.support.RocketMQHeaders;
|
||||
import org.springframework.messaging.Message;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
@RocketMQTransactionListener(txProducerGroup = "myTxProducerGroup", corePoolSize = 5, maximumPoolSize = 10)
|
||||
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {
|
||||
@Override
|
||||
public RocketMQLocalTransactionState executeLocalTransaction(Message msg,
|
||||
Object arg) {
|
||||
Object num = ((HashMap) msg.getHeaders().get(RocketMQHeaders.PROPERTIES))
|
||||
.get("USERS_test");
|
||||
|
||||
if ("1".equals(num)) {
|
||||
System.out.println(
|
||||
"executer: " + new String((byte[]) msg.getPayload()) + " unknown");
|
||||
return RocketMQLocalTransactionState.UNKNOWN;
|
||||
}
|
||||
else if ("2".equals(num)) {
|
||||
System.out.println(
|
||||
"executer: " + new String((byte[]) msg.getPayload()) + " rollback");
|
||||
return RocketMQLocalTransactionState.ROLLBACK;
|
||||
}
|
||||
System.out.println(
|
||||
"executer: " + new String((byte[]) msg.getPayload()) + " commit");
|
||||
return RocketMQLocalTransactionState.COMMIT;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
|
||||
System.out.println("check: " + new String((byte[]) msg.getPayload()));
|
||||
return RocketMQLocalTransactionState.COMMIT;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
logging.level.org.springframework.cloud.stream.binder.rocketmq=DEBUG
|
||||
|
||||
spring.cloud.stream.rocketmq.binder.name-server=127.0.0.1:9876
|
||||
|
||||
spring.cloud.stream.bindings.output1.destination=test-topic
|
||||
spring.cloud.stream.bindings.output1.content-type=application/json
|
||||
spring.cloud.stream.rocketmq.bindings.output1.producer.group=binder-group
|
||||
spring.cloud.stream.rocketmq.bindings.output1.producer.sync=true
|
||||
|
||||
spring.cloud.stream.bindings.output2.destination=TransactionTopic
|
||||
spring.cloud.stream.bindings.output2.content-type=application/json
|
||||
spring.cloud.stream.rocketmq.bindings.output2.producer.transactional=true
|
||||
spring.cloud.stream.rocketmq.bindings.output2.producer.group=myTxProducerGroup
|
||||
|
||||
spring.application.name=rocketmq-produce-example
|
||||
|
||||
server.port=28081
|
||||
|
||||
management.endpoints.web.exposure.include=*
|
||||
management.endpoint.health.show-details=always
|
||||
Reference in New Issue
Block a user