<repo_name>madwenoma/elasticsearch-demos<file_sep>/src/main/java/com/lee/learn/house/domain/House.java
package com.lee.learn.house.domain;

public class House {
    private Long houseId;
    private String title;
    private String xiaoQu;
    private Double price;
    private Integer junJia;
    private String jingJiRen;
    private String huXing;
    private String louCeng;
    private Double jianZhuMianJi;
    private String huXingJieGou;
    private Double shiNeiMianJi;
    private String jianZhuLeiXing;
    private String chaoXiang;
    private String jianZhuJieGou;
    private String zhuangXiu;
    private String tiHuBiLi;
    private String gongNuanFangShi;
    private String dianTi;
    private String chanQuan;

    public Long getHouseId() {
        return houseId;
    }

    public void setHouseId(Long houseId) {
        this.houseId = houseId;
    }

    public String getTitle() {
        return title;
    }

    public void setTitle(String title) {
        this.title = title;
    }

    public String getXiaoQu() {
        return xiaoQu;
    }

    public void setXiaoQu(String xiaoQu) {
        this.xiaoQu = xiaoQu;
    }

    public Double getPrice() {
        return price;
    }

    public void setPrice(Double price) {
        this.price = price;
    }

    public Integer getJunJia() {
        return junJia;
    }

    public void setJunJia(Integer junJia) {
        this.junJia = junJia;
    }

    public String getJingJiRen() {
        return jingJiRen;
    }

    public void setJingJiRen(String jingJiRen) {
        this.jingJiRen = jingJiRen;
    }

    public String getHuXing() {
        return huXing;
    }

    public void setHuXing(String huXing) {
        this.huXing = huXing;
    }

    public String getLouCeng() {
        return louCeng;
    }

    public void setLouCeng(String louCeng) {
        this.louCeng = louCeng;
    }

    public Double getJianZhuMianJi() {
        return jianZhuMianJi;
    }

    public void setJianZhuMianJi(Double jianZhuMianJi) {
        this.jianZhuMianJi = jianZhuMianJi;
    }

    public String getHuXingJieGou() {
        return huXingJieGou;
    }

    public void setHuXingJieGou(String huXingJieGou) {
        this.huXingJieGou = huXingJieGou;
    }

    public Double getShiNeiMianJi() {
        return shiNeiMianJi;
    }

    public void setShiNeiMianJi(Double shiNeiMianJi) {
        this.shiNeiMianJi = shiNeiMianJi;
    }

    public String getJianZhuLeiXing() {
        return jianZhuLeiXing;
    }

    public void setJianZhuLeiXing(String jianZhuLeiXing) {
        this.jianZhuLeiXing = jianZhuLeiXing;
    }

    public String getChaoXiang() {
        return chaoXiang;
    }

    public void setChaoXiang(String chaoXiang) {
        this.chaoXiang = chaoXiang;
    }

    public String getJianZhuJieGou() {
        return jianZhuJieGou;
    }

    public void setJianZhuJieGou(String jianZhuJieGou) {
        this.jianZhuJieGou = jianZhuJieGou;
    }

    public String getZhuangXiu() {
        return zhuangXiu;
    }

    public void setZhuangXiu(String zhuangXiu) {
        this.zhuangXiu = zhuangXiu;
    }

    public String getTiHuBiLi() {
        return tiHuBiLi;
    }

    public void setTiHuBiLi(String tiHuBiLi) {
        this.tiHuBiLi = tiHuBiLi;
    }

    public String getGongNuanFangShi() {
        return gongNuanFangShi;
    }

    public void setGongNuanFangShi(String gongNuanFangShi) {
        this.gongNuanFangShi = gongNuanFangShi;
    }

    public String getDianTi() {
        return dianTi;
    }

    public void setDianTi(String dianTi) {
        this.dianTi = dianTi;
    }

    public String getChanQuan() {
        return chanQuan;
    }

    public void setChanQuan(String chanQuan) {
        this.chanQuan = chanQuan;
    }

    @Override
    public String toString() {
        final StringBuilder sb = new StringBuilder("House{");
        sb.append("title='").append(title).append('\'');
        sb.append(", xiaoQu='").append(xiaoQu).append('\'');
        sb.append(", price=").append(price);
        sb.append(", junJia=").append(junJia);
        sb.append(", jingJiRen='").append(jingJiRen).append('\'');
        sb.append(", huXing='").append(huXing).append('\'');
        sb.append(", louCeng='").append(louCeng).append('\'');
        sb.append(", jianZhuMianJi=").append(jianZhuMianJi);
        sb.append(", huXingJieGou='").append(huXingJieGou).append('\'');
        sb.append(", shiNeiMianJi=").append(shiNeiMianJi);
        sb.append(", jianZhuLeiXing='").append(jianZhuLeiXing).append('\'');
        sb.append(", chaoXiang='").append(chaoXiang).append('\'');
        sb.append(", jianZhuJieGou='").append(jianZhuJieGou).append('\'');
        sb.append(", zhuangXiu='").append(zhuangXiu).append('\'');
        sb.append(", tiHuBiLi='").append(tiHuBiLi).append('\'');
        sb.append(", gongNuanFangShi='").append(gongNuanFangShi).append('\'');
        sb.append(", dianTi='").append(dianTi).append('\'');
        sb.append(", chanQuan='").append(chanQuan).append('\'');
        sb.append('}');
        return sb.toString();
    }
}<file_sep>/src/main/java/com/lee/learn/douban/MoviePage.java
package com.lee.learn.douban;

import java.util.List;

public class MoviePage {

    private List<Movie> subjects;

    public List<Movie> getSubjects() {
        return subjects;
    }

    public void setSubjects(List<Movie> subjects) {
        this.subjects = subjects;
    }

}
<file_sep>/src/main/java/com/lee/learn/house/rabbit/demos/DeadLetterDemoSender.java
package com.lee.learn.house.rabbit.demos;

import com.lee.learn.RabbitMQConfig;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.UUID;

@Component
public class DeadLetterDemoSender {

    @Autowired
    RabbitTemplate rabbitTemplate;

    public void sendDeadLetterMsg(String msg) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        // 声明消息处理器  这个对消息进行处理  可以设置一些参数   对消息进行一些定制化处理
        MessagePostProcessor messagePostProcessor = message -> {
            org.springframework.amqp.core.MessageProperties messageProperties = message.getMessageProperties();
            messageProperties.setContentEncoding("utf-8");
//            设置过期时间10*1000毫秒
//            messageProperties.setExpiration("10000");
            return message;
        };
        //向延迟队列发送消息
        rabbitTemplate.convertAndSend(RabbitMQConfig.delayExchange
                , RabbitMQConfig.delayRoutingKey, msg, messagePostProcessor, correlationData);
    }
}
<file_sep>/src/main/java/com/lee/learn/house/rabbit/demos/transaction/BarService.java
package com.lee.learn.house.rabbit.demos.transaction;

import com.lee.learn.RabbitMQConfig;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.UUID;

//模拟另一个机器上的服务
@Service
public class BarService {

    @Autowired
    private JdbcTemplate jdbcTemplate;//
    @Autowired
    private EventManager eventManager;


    @Transactional
    @RabbitListener(queues = RabbitMQConfig.TRAN_FOO_SUCC_QUEUE)
    public void handleFooSuccEvent(Event event) {//监听队列对象直接转为java bean
        String barId = UUID.randomUUID().toString();
        String barName = "bar1";
        try {
           jdbcTemplate.update("INSERT INTO bar(id, name) values(?,?)", barId, barName);

            //抛出异常，模拟分布式事务失败
            throw new RuntimeException();

        }catch (Exception e) {
            eventManager.sendEventQueue(RabbitMQConfig.TRAN_BAR_FAIL_QUEUE,event);
            throw new AmqpRejectAndDontRequeueException(e);
        }
    }


}
<file_sep>/src/main/java/com/lee/learn/AppStartEvent.java
package com.lee.learn;

import com.lee.learn.douban.DoubanMovieCrawler;
import com.lee.learn.house.crawler.LianjiaCrawler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

/**
 * Created by Administrator on 2018/7/17.
 */
@Component
public class AppStartEvent implements ApplicationListener<ContextRefreshedEvent> {

    @Autowired
    private LianjiaCrawler lianjiaCrawler;
    @Autowired
    private DoubanMovieCrawler doubanMovieCrawler;

    @Override
    @Async
    public void onApplicationEvent(ContextRefreshedEvent event) {
        if (event.getApplicationContext().getParent() == null) {
            System.out.println("onApplicationEvent......");
//            try {
//                lianjiaCrawler.start(2);
//            } catch (Exception e) {
//                e.printStackTrace();
//            }

        }
        new Thread(() -> {
            try {
                doubanMovieCrawler.start(3);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
    }

}
<file_sep>/src/main/java/com/lee/learn/house/rabbit/demos/RabbitRpcClient.java
package com.lee.learn.house.rabbit.demos;

import com.lee.learn.RabbitMQConfig;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class RabbitRpcClient {

    @Autowired
    RabbitTemplate rabbitTemplate;

    public String requestHello(String msg) {
        return (String) rabbitTemplate.convertSendAndReceive(RabbitMQConfig.RPC_EXCHANGE,
                RabbitMQConfig.RPC_ROUTINGKEY, msg);
    }

}
<file_sep>/src/test/java/com/lee/learn/disruptor/DisruptorMain.java
package com.lee.learn.disruptor;

/**
 * @description disruptor代码样例。每10ms向disruptor中插入一个元素，消费者读取数据，并打印到终端
 */

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.ThreadFactory;

public class DisruptorMain {


    static class ElementHandler implements WorkHandler<Element> {

//        @Override
//        public void onEvent(Element element, long l, boolean b) throws Exception {
//            long threadId = Thread.currentThread().getId();
//            System.out.println("Element: " + threadId + " " + element.get());
//        }

        @Override
        public void onEvent(Element element) {
//            System.out.println(element.get());
            if (element.get() == 999998) {
                System.out.println("999998 end:" + System.currentTimeMillis());
            }
        }
    }
    public static void main(String[] args) throws Exception {

        ThreadFactory threadFactory = new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "simpleThread");
            }
        };
        // RingBuffer生产工厂,初始化RingBuffer的时候使用
        EventFactory<Element> factory = new EventFactory<Element>() {
            @Override
            public Element newInstance() {
                return new Element();
            }
        };
        // 处理Event的handler
//        EventHandler<Element> handler = new ElementHandler();
        // 阻塞策略
        SleepingWaitStrategy strategy = new SleepingWaitStrategy();
        // 指定RingBuffer的大小
        int bufferSize = 1024 * 1024;
        // 创建disruptor，采用单生产者模式
        Disruptor<Element> disruptor = new Disruptor(factory, bufferSize, threadFactory,
                ProducerType.SINGLE, strategy);
        // 设置EventHandler
        ElementHandler[] handlers = new ElementHandler[10];
        for (int i = 0; i < 10; i++) {
            handlers[i] = new ElementHandler();
        }
//        disruptor.handleEventsWith(handlers);
        disruptor.handleEventsWithWorkerPool(handlers);
        // 启动disruptor的线程
        disruptor.start();

        RingBuffer<Element> ringBuffer = disruptor.getRingBuffer();
        long start = System.currentTimeMillis();
        System.out.println("start " + start);
        for (int l = 0; l < 999999; l++) {
            // 获取下一个可用位置的下标
            long sequence = ringBuffer.next();
            try {
                // 返回可用位置的元素
                Element event = ringBuffer.get(sequence);
                // 设置该位置元素的值
                event.set(l);
            } finally {
                ringBuffer.publish(sequence);
            }
//            Thread.sleep(1);
        }

        disruptor.shutdown();

        System.in.read();
    }

    // 队列中的元素
    static class Element {
        private int value;

        public int get() {
            return value;
        }

        public void set(int value) {
            this.value = value;
        }

    }        // 生产者的线程工厂


}<file_sep>/src/main/java/com/lee/learn/house/rabbit/demos/transaction/EventManager.java
package com.lee.learn.house.rabbit.demos.transaction;

import com.lee.learn.RabbitMQConfig;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.jdbc.core.JdbcTemplate;

public class EventManager {
    private static final String INSERT_SQL = "insert into event(id,event_type,module_name,module_id,create_time) values(?,?,?,?,?)";
    private static final String SELECT_SQL = "select model_id from event where event_id = ?";

    private final JdbcTemplate jdbcTemplate;
    private final RabbitTemplate rabbitTemplate;

    public EventManager(JdbcTemplate jdbcTemplate, RabbitTemplate rabbitTemplate) {
        this.jdbcTemplate = jdbcTemplate;
        this.rabbitTemplate = rabbitTemplate;
    }

    public void insertEvent(Event event) {
        jdbcTemplate.update(INSERT_SQL, event.getId(), event.getEventType(),
                event.getModelName(), event.getModelId(), event.getCreateTime());
    }

    public String queryModelId(String eventId) {
        return jdbcTemplate.queryForObject(SELECT_SQL, String.class, eventId);
    }

    public void sendEventQueue(String queue, Event event) {
        this.rabbitTemplate.convertAndSend(queue, event);
    }
}
<file_sep>/src/main/java/com/lee/learn/RabbitMQConfig.java
package com.lee.learn;

import com.lee.learn.house.rabbit.demos.transaction.EventManager;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;

import java.util.HashMap;
import java.util.Map;
//https://blog.csdn.net/m0_37867405/article/details/80793601

//https://blog.csdn.net/itguangit/article/details/80031595
@Configuration
public class RabbitMQConfig {

    public static final String EXCHANGE = "boot-topic-exchange";
    public static final String ROUTINGKEY1 = "boot-routing.update";
    public static final String ROUTINGKEY_DELETE = "boot-routing.delete";
    public static final String ROUTINGKEY_All = "boot-routing.#";
    public static final String QUEUE1 = "spring-queue";
    public static final String QUEUE_ALL = "spring-queue-all";

    public static final String RPC_EXCHANGE = "rpc-exchange";
    public static final String RPC_ROUTINGKEY = "rpc";
    public static final String RPC_QUEUE = "rpc-queue";


    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(EXCHANGE);
    }

    @Bean
    public Queue queue() {
        return new Queue(QUEUE1, false);
    }

    @Bean
    public Queue queueForAll() {
        return new Queue(QUEUE_ALL, false);
    }

    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue()).to(topicExchange()).with(ROUTINGKEY1);
    }

    @Bean
    public Binding bindingDel() {
        return BindingBuilder.bind(queue()).to(topicExchange()).with(ROUTINGKEY_DELETE);
    }

    @Bean
    public Binding bindingAll() {
        return BindingBuilder.bind(queueForAll()).to(topicExchange()).with(ROUTINGKEY_All);
    }


    ///////////////////////////////////Rpc 相关 start/////////////////////////////////////

    @Bean
    public DirectExchange rpcExchanger() {
        return new DirectExchange(RPC_EXCHANGE);
    }

    @Bean
    public Queue rpcQueue() {
        return new Queue(RPC_QUEUE, false);
    }

    @Bean
    public Binding rpcBinding() {
        return BindingBuilder.bind(rpcQueue()).to(rpcExchanger()).with(RPC_ROUTINGKEY);
    }

    ///////////////////////////////////Rpc 相关   end/////////////////////////////////////


    ///////////////////////////////////死信队列 相关 start/////////////////////////////////////

    public static String delayQueue = "delay_queue";
    public static String delayExchange = "delay_exchange";
    public static String delayRoutingKey = "delay_routing_key";

    public static final String deadLetterQueue = "dead_letter_queue";
    public static String deadLetterExchange = "dead_letter_exchange";
    public static String deadLetterRoutingKey = "dead_letter_routing_key";

    @Bean
    public DirectExchange delayExchange() {
        DirectExchange ret = new DirectExchange(delayExchange, true, false);
        ret.setInternal(false);
        return ret;
    }

    @Bean
    public DirectExchange deadLetterExchange() {
        return new DirectExchange(deadLetterExchange, true, false);
    }

    /**
     * 延迟队列没有consumer,只是作为向dead letter queue转发消息的存储中介
     * 消息开始是发送给延迟队列，30s后被转发向死信队列，消费者监听死信队列收到消息，实现定时延迟消息。
     */
    @Bean
    public Queue delayQueue() {
        Map<String, Object> arguments = new HashMap<String, Object>();
        arguments.put("x-message-ttl", 30 * 1000); // Message TTL = 30s
        arguments.put("x-max-length", 1000000); // Max length = 1million
        // 死信路由到死信交换器DLX
        arguments.put("x-dead-letter-exchange", deadLetterExchange);
        arguments.put("x-dead-letter-routing-key", deadLetterRoutingKey);
        return new Queue(delayQueue, true, false, false, arguments);
    }

    @Bean
    public Queue deadLetterQueue() {
        return new Queue(deadLetterQueue, true, false, false);
    }

    @Bean
    public Binding bindingDelay() {
        return BindingBuilder.bind(delayQueue()).to(delayExchange()).with(delayRoutingKey);
    }

    @Bean
    public Binding bindingDeadLetter() {
        return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with(deadLetterRoutingKey);
    }

    ///////////////////////////////////死信队列 相关   end/////////////////////////////////////


    ////////////////////////////////分布式事务相关 start/////////////////////////////////

    public static final String TRAN_FOO_SUCC_QUEUE = "foo-success-queue";
    public static final String TRAN_BAR_FAIL_QUEUE = "bar-failure-queue";

    @Bean
    public Queue successQueue() {
        return new Queue(TRAN_FOO_SUCC_QUEUE);
    }

    @Bean
    public Queue failureQueue() {
        return new Queue(TRAN_BAR_FAIL_QUEUE);
    }

    @Bean
    public Jackson2JsonMessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public EventManager eventManager(JdbcTemplate jdbcTemplate, RabbitTemplate rabbitTemplate) {
        return new EventManager(jdbcTemplate, rabbitTemplate);
    }

    ////////////////////////////////分布式事务相关 end/////////////////////////////////


//    @Bean
//    public SimpleMessageListenerContainer messageContainer() {
//        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
//        container.setQueues(queue());
//        container.setExposeListenerChannel(true);
//        container.setMaxConcurrentConsumers(1);
//        container.setConcurrentConsumers(1);
//        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认
//        container.setMessageListener(new ChannelAwareMessageListener() {
//            @Override
//            public void onMessage(Message message, Channel channel) throws Exception {
//                byte[] body = message.getBody();
//                System.out.println("receive msg queue: " + new String(body));
//                Thread.sleep(10000);
//
//                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //确认消息成功消费
//
//            }
//        });
//        return container;
//    }
}
<file_sep>/src/main/java/com/lee/learn/house/rabbit/demos/transaction/event.sql
CREATE TABLE event (
  `id` char(36) NOT NULL DEFAULT '',
  `event_type` varchar(100) DEFAULT NULL,
  `model_name` varchar(100) DEFAULT NULL,
  `model_id` char(36) DEFAULT NULL ,
  `create_time` bigint(11) DEFAULT NULL ,
  PRIMARY KEY (`id`)
)ENGINE=InnoDB DEFAULT CHARSET=utf8

CREATE TABLE foo (
  `id` char(36) NOT NULL DEFAULT '',
  `name` varchar(100) DEFAULT NULL,
  PRIMARY KEY (`id`)
)ENGINE=InnoDB DEFAULT CHARSET=utf8


-- 在另一个库上执行
CREATE TABLE bar (
  `id` char(36) NOT NULL DEFAULT '',
  `name` varchar(100) DEFAULT NULL,
  PRIMARY KEY (`id`)
)ENGINE=InnoDB DEFAULT CHARSET=utf8<file_sep>/src/test/java/com/lee/learn/house/HouseIndexOperatorTest.java
package com.lee.learn.house;

import com.lee.learn.house.crawler.LianjiaCrawler;
import com.lee.learn.house.domain.House;
import com.lee.learn.house.domain.HouseIndexTemplate;
import junit.framework.TestCase;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

/**
 * Created by Administrator on 2018/7/16.
 */
@RunWith(SpringRunner.class)
@SpringBootTest
public class HouseIndexOperatorTest extends TestCase {

    @Autowired
    HouseIndexOperator operator;

    @Test
    public void testCreateIndex() throws Exception {
        HouseIndexTemplate house = new HouseIndexTemplate();
        house.setHouseId(1L);
        house.setJingJiRen("xiaowang");
        house.setJunJia(42333);
        house.setPrice(324.0);
        house.setTitle("月季园 通透");
        house.setXiaoQu("月季园");
        house.setChanQuan("70年");
        house.setChaoXiang("南 北");
        house.setDianTi("有");
        house.setGongNuanFangShi("地热");
        house.setHuXing("连体一户");
        house.setJianZhuJieGou("");
        house.setJianZhuLeiXing("");
        house.setJianZhuMianJi(40.0);
        house.setShiNeiMianJi(20.0);
        System.out.println(operator.createIndex(house));
    }


}