lagou 爪哇 3-3 dubbo 笔记

cd /Users/ale/Downloads/apache-zookeeper-3.5.8-bin/bin
./zkServer.sh start-foreground  
./zkServer.sh start
./zkServer.sh status 
./zkServer.sh stop

1. 从git 上下载项目
https://github.com/apache/dubbo-admin 
或者直接下载zip包
https://codeload.github.com/apache/dubbo-admin/zip/master
2. 在 dubbo-admin-server/src/main/resources/application.properties
# 这里可以修改 tomcat 默认端口, 可选
server.port=7001
# 登录窗口的用户名和密码
spring.root.password=root
spring.guest.password=guest
# 指定注册中心地址, 这里根据自己的实际需要进行更改
dubbo.registry.address=zookeeper://127.0.0.1:2181
3.切换到项目所在的路径 使用mvn 打包 
mvn clean package -Dmaven.test.skip=true
4.java 命令运行
java -jar 对应的jar包
java -jar ./target/dubbo-admin-0.0.1-SNAPSHOT.jar
5. 访问 http://localhost:7001
使用用户名root 和 密码root 进行登录即可.

<!-- 消费方应用名，用于计算依赖关系，不是匹配条件，不要与提供方一样 -->
    <dubbo:application name="consumer-of-helloworld-app" owner="chengxu">
        <dubbo:parameter key="qos.enable" value="true"></dubbo:parameter>
        <dubbo:parameter key="qos.port" value="22223"></dubbo:parameter>
        <!--默认是 false-->
        <dubbo:parameter key="qos.accept.foreign.ip" value="false"></dubbo:parameter>
    </dubbo:application>

# 这里写上应用的名字
dubbo.application.name=service-consumer
dubbo.application.qosEnable=true
dubbo.application.qosPort=true
dubbo.application.qosAcceptForeignIp=true

com.lagou.service.impl.HumanHelloService

package com.lagou.service.impl;
import com.lagou.service.HelloService;
/**
 * spi：人类打招呼
 */
public class HumanHelloService implements HelloService {
    @Override
    public String sayHello(String name) {
        return "hello my friend";
    }
}

package com.lagou.test;
import com.lagou.service.HelloService;
import java.util.ServiceLoader;
public class JavaSpiMain {
    public static void main(String[] args) {
        // 怎么得到实现了 HelloService 接口的组件呢，这里用到了 ServiceLoader 类
        final ServiceLoader<HelloService> list = ServiceLoader.load(HelloService.class);
        for (HelloService helloService : list) {
            System.out.println("类名：" + helloService.getClass().getName());
            System.out.println(helloService.sayHello("hi！"));
        }
    }
}

package com.lagou.service;
import org.apache.dubbo.common.extension.SPI;
@SPI
public interface HelloServiceForDubbo {
    String sayHello(String name);
}

package com.lagou.test;
import com.lagou.service.HelloServiceForDubbo;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;
import java.util.Set;
public class DubboSpiMain {
    public static void main(String[] args) {
        // 获取拓展加载器
        final ExtensionLoader<HelloServiceForDubbo> extensionLoader =
                ExtensionLoader.getExtensionLoader(HelloServiceForDubbo.class);
        // 遍历所有支持的拓展点 META-INF/dubbo
        final Set<String> extensionNames = extensionLoader.getSupportedExtensions();
        for (String extensionName : extensionNames) {
            // 遍历拓展点对象
            final HelloServiceForDubbo helloServiceForDubbo = extensionLoader.getExtension(extensionName);
            System.out.println(helloServiceForDubbo.sayHello(""));
    }
}

public class DubboAdaptiveMain {
  public static void main(String[] args) {
    URL url = URL.valueOf("test://localhost/hello?hello.service=dog");
    final HelloService adaptiveExtension = ExtensionLoader.getExtensionLoader(HelloService.class).getAdaptiveExtension();
    adaptiveExtension.sayHello(url);
  }
}

HelloService.class -> hello.service

package com.lagou.service;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.Adaptive;
import org.apache.dubbo.common.extension.SPI;
/**
 * 指定默认匹配 human
 */
@SPI("human")
public interface HelloServiceForDubbo {
    /**
     * 这里加上 Adaptive 注解，使得有动态的选择能力
     * @param url
     * @return
     */
    @Adaptive
    String sayHello(URL url);
}

package com.lagou;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.threadpool.ThreadPool;
import org.apache.dubbo.common.threadpool.support.fixed.FixedThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.concurrent.*;
/**
 * @author likai
 * @date 2020/11/21
 */
public class WatchThreadPool extends FixedThreadPool implements Runnable {
    private static final Logger log = LoggerFactory.getLogger(WatchThreadPool.class);
    // 定义一个阈值
    private static final double ALARM_PERCENT = 0.80;
    private final Map<URL, ThreadPoolExecutor> THREAD_POOLS = new ConcurrentHashMap<>();
    // 每隔3秒打印线程使用情况
    public WatchThreadPool() {
        Executors.newSingleThreadScheduledExecutor()
                .scheduleWithFixedDelay(this, 1, 3, TimeUnit.SECONDS);
    }
    @Override
    public Executor getExecutor(URL url) {
        // 从父类中创建线程池
        final Executor executor = super.getExecutor(url);
        if (executor instanceof ThreadPoolExecutor) {
            THREAD_POOLS.put(url, ((ThreadPoolExecutor) executor));
        }
        return executor;
    }
    @Override
    public void run() {
        // 遍历线程池，如果超出指定的部分，进行操作，比如接入公司的告警系统或者短信平台
        for (Map.Entry<URL, ThreadPoolExecutor> entry : THREAD_POOLS.entrySet()) {
            final URL url = entry.getKey();
            final ThreadPoolExecutor executor = entry.getValue();
            // 当前执行中的线程数
            final int activeCount = executor.getActiveCount();
            // 总计线程数
            final int poolSize = executor.getCorePoolSize();
            double used = (double) activeCount / poolSize;
            final int usedNum = (int) (used * 100);
            log.info("线程池执行状态:[{}/{}]:{}%", activeCount, poolSize, usedNum);
            if (used >= ALARM_PERCENT) {
                log.error("超出警戒值！host:{}, 当前已使用量:{}%, URL:{}",
                        url.getIp(), usedNum, url);
            }
        }
    }
}

watching=包名.线程池名

dubbo.provider.threadpool=watching

public class DubboRouterMain {
  public static void main(String[] args) {
    RegistryFactory registryFactory =
ExtensionLoader.getExtensionLoader(RegistryFactory.class).getAdaptiveExtension();
Registry registry = registryFactory.getRegistry(URL.valueOf("zookeeper://127.0.0.1:2181"));
registry.register(URL.valueOf("condition://0.0.0.0/com.lagou.service.HelloService?category=routers&force=true&dynamic=true&rule=" + URL.encode("=> host != 你的机器ip不能是127.0.0.1")));
  }
}

restartInstances=com.lagou.router.RestartingInstanceRouterFactory

ReadyRestartInstances.create().addRestartingInstance("service-provider", "正在重新启动的机器IP");

public interface HelloService {
    String sayHello(String name);
}

@Service(interfaceClass = HelloService.class,retries = 0)
@Component
public class HelloServiceImpl implements HelloService {
    @Override
    public String sayHello(String name) {
        String requestIp = RpcContext.getContext().getAttachment(Constants.REQUEST_IP);
        System.out.println("provider1: sayHello + " +requestIp);
        return "provider1: sayHello + " +requestIp;
    }
}

transportIPFilter=com.lagou.filter.TransportIPFilter

/**
 * 需要加入 注解 或者 手动开启，否则过滤器不生效
 */
@Component
@WebFilter(urlPatterns = "/hello")
public class IPFilter implements Filter {
    @Override
    public void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain)
            throws IOException, ServletException {
        try {
            final String remoteAddr = servletRequest.getRemoteAddr();
            System.out.println("过滤器 doFilter 方法获得 remoteAddr = " + remoteAddr);
            IPThreadLocal.set(servletRequest.getRemoteAddr());
            filterChain.doFilter(servletRequest, servletResponse);
        } finally {
            IPThreadLocal.remove();
        }
    }
}

/**
 * 给过滤器起一个名字
 */
@Activate
public class TransportIPFilter implements Filter {
    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        // 获得 IP
        final String ip = IPThreadLocal.get();
        RpcContext.getContext().getAttachments().put(Constants.REQUEST_IP, ip);
        System.out.println("dubbo Filter 存入ip" + ip);
        return invoker.invoke(invocation);
    }
}

public class IPThreadLocal {
    private static final ThreadLocal<String> THREAD_LOCAL = new ThreadLocal<String>();
    public static void set(String ip) {
        THREAD_LOCAL.set(ip);
    }
    public static String get() {
        return THREAD_LOCAL.get();
    }
    public static void remove() {
        THREAD_LOCAL.remove();
    }
}

@RestController
public class HelloController {
    @Reference(loadbalance = "roundrobin")
    private HelloService dubboTestApi;
    @GetMapping("/hello")
    public String hello() {
        return this.dubboTestApi.sayHello("dubbo");
    }
}

/**
 * 定义一个基础接口
 */
public interface HelloService {
    String methodA(String name, int mills);
    String methodB(String name, int mills);
    String methodC(String name, int mills);
}

public class HelloServiceImpl implements HelloService {
    @Override
    public String methodA(String name, int mills) {
        try {
            TimeUnit.MILLISECONDS.sleep(mills);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("method A : " + name);
        return "method A : " + name;
    }
    @Override
    public String methodB(String name, int mills) {
        try {
            TimeUnit.MILLISECONDS.sleep(mills);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("method B : " + name);
        return "method B : " + name;
    }
    @Override
    public String methodC(String name, int mills) {
        try {
            TimeUnit.MILLISECONDS.sleep(mills);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("method C : " + name);
        return "method C : " + name;
    }
}

ExecutorService executorService = Executors.newFixedThreadPool(3);
        // 确保在1分钟内可以被调用 2000 次以上
        HelloService helloService = (HelloService) context.getBean("helloService");
        for (int i = 0; i < 2300; i++) {
            executorService.execute(new HelloServiceThread(helloService));
        }

class HelloServiceThread extends Thread {
    private final HelloService helloService;
    HelloServiceThread(HelloService helloService) {
        super();
        this.helloService = helloService;
    }
    @Override
    public void run() {
        try {
            TimeUnit.MILLISECONDS.sleep(30);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        final int i = ThreadLocalRandom.current().nextInt(3);
        String result = null;
        switch (i) {
            case 0:
                result = helloService.methodA("", ThreadLocalRandom.current().nextInt(100));
                break;
            case 1:
                result = helloService.methodB("", ThreadLocalRandom.current().nextInt(100));
                break;
            case 2:
                result = helloService.methodC("", ThreadLocalRandom.current().nextInt(100));
                break;
            default:
                break;
        }
        // 调用结果
        System.out.println(Thread.currentThread().getName() + " result = " + Objects.toString(result, ""));
    }
}

public class XmlConsumer {
    public static void main(String[] args) throws Exception {
        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"classpath:dubbo-demo-consumer.xml"});
        // 起始时间
        long startMills = Instant.now().toEpochMilli();
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        // 确保在1分钟内可以被调用 2000 次以上
        HelloService helloService = (HelloService) context.getBean("helloService");
        for (int i = 0; i < 2300; i++) {
            executorService.execute(new HelloServiceThread(helloService));
        }
        executorService.shutdown();
        executorService.awaitTermination(3, TimeUnit.MINUTES);
        // 终止时间
        long endMills = Instant.now().toEpochMilli();
        System.out.println("-----总耗时" + (endMills - startMills) + "-----");
    }
}

@Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        long epochMilli = Instant.now().toEpochMilli();
        Result invoke = invoker.invoke(invocation);
        final LocalDateTime now = LocalDateTime.now();
        final Instant endInstant = now.atZone(ZoneId.systemDefault()).toInstant();
        // 方法耗时
        int costMills = (int) (endInstant.toEpochMilli() - epochMilli);
        // 方法名称
        String methodName = invocation.getMethodName();
        ...
        ...

/**
         * recordArray 共计 60 等份
         *
         *      调用时间 1ms 2ms 3ms 4ms 5ms
         * 方法methodA   1   2   3
         * 方法methodB   3   3   1       5
         * 方法methodC   1       8   3
         */
        // localDateTime 对象包含了时分秒信息，取出秒 作为下标
        int index = now.getSecond();
        // 这里肯定不为 null, 若时间比对成功则则 进行 put
        if (recordArray[index].isDateTimeEquals(now)) {
            Integer value = recordArray[index].get(methodName, costMills);
            if (value != null) {
                recordArray[index].put(methodName, costMills, value + 1);
            } else {
                recordArray[index].put(methodName, costMills, 1);
            }
        } else {
            // 否则表示钟或小时至少有一个不存在则清空该 table 后重新put
            recordArray[index].clear();
            recordArray[index].setDateTime(now);
            // 秒不相等，判断分是否相等。 若相等。
            // 0~1 1_2 58~59 59~60(0)  一看到1分0秒，就表示完结了。   3分1秒    4分0秒       9分3秒 ~ 10分2秒
            // 0             59                                   181~182  240~241      543~544 ~ 602~603
            recordArray[index].put(methodName, costMills, 1);
        }

/**
 * 一种类似表格的数据接口，类似 Map<A, Map<B,C>>类型。
 */
class RecordTable {
    /**
     * 总记录数
     */
    private static int count = 0;
    /**
     * 内部使用线程安全的 Google guava 包下的 Table 类型
     */
    private final Table<String, Integer, Integer> table = Tables.synchronizedTable(HashBasedTable.create());
    /**
     * 包含时间信息，记录为 1 秒
     */
    private LocalDateTime dateTime;
    public static RecordTable[] newArray() {
        count = 0;
        // 共计 60 等份
        int length = 60;
        final RecordTable[] array = new RecordTable[length];
        final LocalDateTime now = LocalDateTime.now();
        for (int i = 0; i < 60; i++) {
            array[i] = new RecordTable();
            array[i].setDateTime(now);
        };
        return array;
    }
    public RecordTable() {}
    public Integer get(@Nullable Object rowKey, @Nullable Object columnKey) {
        return table.get(rowKey, columnKey);
    }
    public Integer put(String rowKey, Integer columnKey, Integer value) {
        count++;
        System.out.println("put 的次数 =" +  count);
        return table.put(rowKey, columnKey, value);
    }
    /**
     * 若 日期 和 时分秒 相等则认为是相等
     *
     * @param obj
     * @return
     */
    public boolean isDateTimeEquals(Object obj) {
        if (obj instanceof LocalDateTime) {
            LocalDateTime other = (LocalDateTime) obj;
            if (this.dateTime != null && this.dateTime.toLocalDate().equals(other.toLocalDate())) {
                final LocalTime localTime = this.dateTime.toLocalTime();
                final LocalTime otherLocalTime = other.toLocalTime();
                return localTime.getHour() == otherLocalTime.getHour()
                        && localTime.getMinute() == otherLocalTime.getMinute()
                        && localTime.getSecond() == otherLocalTime.getSecond();
            }
            return false;
        }
        return false;
    }
    public void clear() {
        table.clear();
    }
    public Map<String, Map<Integer, Integer>> rowMap() {
        return table.rowMap();
    }
    public boolean isEmpty(){
        return table.isEmpty();
    }
    public LocalDateTime getDateTime() {
        return dateTime;
    }
    public void setDateTime(LocalDateTime dateTime) {
        this.dateTime = dateTime;
    }
}

@Override
    public void run() {
        // 每隔 5s 打印一次最近1分钟内每个方法的TP90、TP99的耗时情况
        System.out.println("----------每5s一次的任务开始----------");
        // 总行数（最大值为 length）
        int totalColumnSize = 0;
        // 总记录数(执行方法的总次数)
        int totalSize = 0;
        int aTotalSize = 0;
        int bTotalSize = 0;
        int cTotalSize = 0;
        List<Integer> listA = new ArrayList<>(800);
        List<Integer> listB = new ArrayList<>(800);
        List<Integer> listC = new ArrayList<>(800);
        for (int i = 0; i < recordArray.length; i++) {
            if (recordArray[i] == null || recordArray[i].isEmpty()) {
                continue;
            }
            // 行记录数 = (方法A+B+C调用的总次数)
            int size = 0;
            // 每行方法A的调用次数
            int methodAtotalSize = 0;
            int methodBtotalSize = 0;
            int methodCtotalSize = 0;
            final Map<String, Map<Integer, Integer>> rowMap = recordArray[i].rowMap();
            for (Map.Entry<String, Map<Integer, Integer>> entry : rowMap.entrySet()) {
                final String methodName = entry.getKey();
                final Map<Integer, Integer> value = entry.getValue();
                if (value != null) {
                    for (Map.Entry<Integer, Integer> valueEntry : value.entrySet()) {
                        // 消耗毫秒数
                        Integer costMills = valueEntry.getKey();
                        // 消费次数
                        final Integer count = valueEntry.getValue();
                        size += count;
                        if ("methodA".equalsIgnoreCase(methodName)) {
                            methodAtotalSize += count;
                            for (int j = 0; j < count; j++) {
                                listA.add(costMills);
                            }
                        }
                        else if ("methodB".equalsIgnoreCase(methodName)) {
                            methodBtotalSize += count;
                            for (int j = 0; j < count; j++) {
                                listB.add(costMills);
                            }
                        }
                        else if ("methodC".equalsIgnoreCase(methodName)) {
                            methodCtotalSize += count;
                            for (int j = 0; j < count; j++) {
                                listC.add(costMills);
                            }
                        }
                    }
                    if ("methodA".equalsIgnoreCase(methodName)) {
                        aTotalSize += methodAtotalSize;
                    } else if ("methodB".equalsIgnoreCase(methodName)) {
                        bTotalSize += methodBtotalSize;
                    } else if ("methodC".equalsIgnoreCase(methodName)) {
                        cTotalSize += methodCtotalSize;
                    }
                }
            }
            totalColumnSize++;
            totalSize += size;
            System.out.println("行记录数" + size
                    + "其中A记录数 = " + methodAtotalSize
                    + ", B记录数 = " + methodBtotalSize
                    + ", C记录数 = " + methodCtotalSize
                    + ", 下标为" + i
                    + "， 时间为: " + recordArray[i].getDateTime()
                    + ", " + recordArray[i].rowMap());
        }
        System.out.println("总记录数 = " + totalSize + "("+ aTotalSize +", " + bTotalSize  +", " + cTotalSize +")"
                + ", 总行数 = " + totalColumnSize + "\n------每5s一次的任务结束-----");
        // A的 top 90
        int aTop90 = (int) Math.ceil(aTotalSize * 0.9D) -1;
        int aTop99 = (int) Math.ceil(aTotalSize * 0.99D) - 1;
        // 升序取第 %90 位 或 逆序取第 %10
        System.out.println("top A T90下标=" + aTop90 + ", T99 下标=" + aTop99);
        Object[] a = listA.toArray();
        Arrays.sort(a, (Comparator) new TPComparator());
        System.out.println(Arrays.toString(a));
        System.out.println("top A T90 =" + a[aTop90] + ", T99 =" + a[aTop99] + "\n---");
        // B的 top 90
        int bTop90 = (int) Math.ceil(bTotalSize * 90D / 100) - 1;
        int bTop99 = (int) Math.ceil(bTotalSize * 99D / 100) - 1;
        System.out.println("top B T90下标=" + bTop90 + ", T99下标=" + bTop99);
        Object[] b = listB.toArray();
        Arrays.sort(b, (Comparator) new TPComparator());
        System.out.println(Arrays.toString(b));
        System.out.println("top B T90 =" + b[bTop90] + ", T99 =" + b[bTop99] + "\n---");
        // C的 top 90
        int cTop90 = (int) Math.ceil(cTotalSize * 90D / 100) - 1;
        int cTop99 = (int) Math.ceil(cTotalSize * 99D / 100) - 1;
        System.out.println("top C T90下标=" + cTop90 + ", T99下标=" + cTop99);
        Object[] c = listC.toArray();
        Arrays.sort(c, (Comparator) new TPComparator());
        System.out.println(Arrays.toString(c));
        System.out.println("top C T90 =" + c[cTop90] + ", T99 =" + c[cTop99] + "\n---");
    }
}