幂等解决方案集合（二）消息幂等

select * from t_order where order_no = 'order123'
if(order  != null) {
    return ;//消息重复，直接返回
}

<dependency>
    <groupId>com.aliyun.openservices</groupId>
    <artifactId>ons-client-ext</artifactId>
    <version>1.8.4.Final</version>
</dependency>
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-context</artifactId>
    <version>${spring-version}</version>
</dependency>
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-jdbc</artifactId>
    <version>${spring-version}</version>
</dependency>

CREATE TABLE `transaction_record` (
  `consumer_group` varchar(128) NOT NULL DEFAULT '',
  `message_id` varchar(255) NOT NULL DEFAULT '',
  `topic_name` varchar(255) NOT NULL DEFAULT '',
  `ctime` bigint(20) NOT NULL,
  `queue_id` int(11) NOT NULL,
  `offset` bigint(20) NOT NULL,
  `broker_name` varchar(255) NOT NULL DEFAULT '',
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  PRIMARY KEY (`id`),
  UNIQUE KEY `message_id_unique_key` (`message_id`),
  KEY `ctime_key` (`ctime`),
  KEY `load_key` (`queue_id`,`broker_name`,`topic_name`,`ctime`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

/**
 * TestExactlyOnceProducer 启动
 * 通过 PropertyKeyConst.EXACTLYONCE_DELIVERY 开启 exactly-once 投递语义。
 */
public class TestExactlyOnceProducer {
    public static void main(String[] args) {
        Properties producerProperties = new Properties();
        producerProperties.setProperty(PropertyKeyConst.GROUP_ID,"{GROUP_ID}");
        producerProperties.setProperty(PropertyKeyConst.AccessKey,"{accessKey}");
        producerProperties.setProperty(PropertyKeyConst.SecretKey,"{secretKey}");
        producerProperties.setProperty(PropertyKeyConst.NAMESRV_ADDR,"{NAMESRV_ADDR}");
        producerProperties.setProperty(PropertyKeyConst.EXACTLYONCE_DELIVERY,"true");
        Producer producer = ExactlyOnceONSFactory.createProducer(producerProperties);
        producer.start();
        System.out.println("Producer Started");
        for (int i = 0; i < 10; i++) {
            Message message = new Message("{topic}", "{tag}", "mq send transaction message test".getBytes());
            try {
                SendResult sendResult = producer.send(message);
                assert sendResult != null;
                System.out.println(new Date() + " Send mq message success! msgId is: " + sendResult.getMessageId());
            } catch (ONSClientException e) {
                System.out.println("发送失败");
                //出现异常意味着发送失败，为避免消息丢失，建议缓存该消息然后进行重试。
            }
        }
        producer.shutdown();
    }
}

/**
 * TestExactlyOnceListener 实现。
 * 实现了一个事务中对多个业务表进行更新的场景，保证事务内的操作有且仅有一次生效。
 */
public class SimpleTxListener implements MessageListener {
    private MQDataSource dataSource;
    public SimpleTxListener() {
        this.dataSource = new MQDataSource("{url}", "{user}", "{passwd}", "{driver}");
    }
        @Override
    public Action consume(Message message, ConsumeContext context) {
        Connection connection = null;
        Statement statement = null;
        try {
            /**
             *  在此处对消费到的消息 message 做业务计算处理，使用 MQDataSource 将处理结果持久化到数据库系统。
             *  此范例演示了在一个事务内对多个表进行更新的业务场景，Exactly-Once 投递语义保证事务内的操作有且仅有一次。
             *  实际的业务处理按照：接收消息->业务处理->结果持久化的流程来设计。
             */
            connection = dataSource.getConnection();
            connection.setAutoCommit(false);
            String insertSql = String.format("INSERT INTO app(msg, message_id, ctime) VALUES(\"%s\", \"%s\", %d)",
                new String(message.getBody()), message.getMsgID(), System.currentTimeMillis());
            String updateSql = String.format("UPDATE consume_count SET cnt = count + 1 WHERE consumer_group = \"%s\"", "GID_TEST");
            statement = connection.createStatement();
            statement.execute(insertSql);
            statement.execute(updateSql);
            connection.commit();
            System.out.println("consume message :" + message.getMsgID());
            return Action.CommitMessage;
        } catch (Throwable e) {
            try {
                connection.rollback();
            } catch (Exception e1) {
            }
            System.out.println("consume message fail");
            return Action.ReconsumeLater;
        } finally {
            if (statement != null) {
                try {
                    statement.close();
                } catch (Exception e) { }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (Exception e) { }
            }
        }
    }                                                      
}