部署Linux

pom.xml

@@ -24,17 +24,26 @@
             <artifactId>hippo4j-spring-boot-starter</artifactId>
             <version>1.5.0</version>
         </dependency>
-        <dependency>
+<!--        <dependency>-->
-            <groupId>org.springframework.boot</groupId>
+        <!--            <groupId>org.springframework.boot</groupId>-->
-            <artifactId>spring-boot-starter-amqp</artifactId>
+        <!--            <artifactId>spring-boot-starter-amqp</artifactId>-->
-        </dependency>
+        <!--        </dependency>-->
-        <dependency>
+        <!--        <dependency>-->
-            <groupId>cn.hippo4j</groupId>
+        <!--            <groupId>cn.hippo4j</groupId>-->
-            <artifactId>hippo4j-spring-boot-starter-adapter-rabbitmq</artifactId>
+        <!--            <artifactId>hippo4j-spring-boot-starter-adapter-rabbitmq</artifactId>-->
-            <version>1.5.0</version>
+        <!--            <version>1.5.0</version>-->
-        </dependency>
+        <!--        </dependency>-->
     </dependencies>
 
 
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+
 
 </project>

src/main/java/com/mashibing/config/RabbitMQThreadPoolConfig.java

@@ -1,46 +1,35 @@
-package com.mashibing.config;
+//package com.mashibing.config;
-
+//
-import org.springframework.amqp.rabbit.config.AbstractRabbitListenerContainerFactory;
+//import org.springframework.amqp.rabbit.config.AbstractRabbitListenerContainerFactory;
-import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory;
+//import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory;
-import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
+//import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
-import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+//import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Bean;
+//import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Configuration;
+//import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+//
-
+//
-/**
+//
- * @author zjw
+//@Configuration
- * @description
+//public class RabbitMQThreadPoolConfig {
- */
+//
-@Configuration
+//    @Bean
-public class RabbitMQThreadPoolConfig {
+//    public ThreadPoolTaskExecutor rabbitThreadPool(){
-
+//        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
-    /**
+//        executor.setCorePoolSize(5);
-     * 构建消费者要用到的线程
+//        executor.setMaxPoolSize(5);
-     * @return
+//        executor.setQueueCapacity(100);
-     */
+//        executor.setThreadNamePrefix("rabbit-");
-    @Bean
+//        return executor;
-    public ThreadPoolTaskExecutor rabbitThreadPool(){
+//    }
-        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+//
-        executor.setCorePoolSize(5);
+//
-        executor.setMaxPoolSize(5);
+//    @Bean
-        executor.setQueueCapacity(100);
+//    public AbstractRabbitListenerContainerFactory<?> defaultContainerFactory(ThreadPoolTaskExecutor rabbitThreadPool,
-        executor.setThreadNamePrefix("rabbit-");
+//                                                                             AbstractConnectionFactory connectionFactory){
-        return executor;
+//        DirectRabbitListenerContainerFactory factory = new DirectRabbitListenerContainerFactory();
-    }
+//        factory.setConnectionFactory(connectionFactory);
-
+//        connectionFactory.setExecutor(rabbitThreadPool);
-    /**
+//        return factory;
-     * 构建容器工厂，将线程池设置进入
+//    }
-     * @param connectionFactory
+//
-     * @return
+//}
-     */
-    @Bean
-    public AbstractRabbitListenerContainerFactory<?> defaultContainerFactory(ThreadPoolTaskExecutor rabbitThreadPool,
-                                                                             AbstractConnectionFactory connectionFactory){
-        DirectRabbitListenerContainerFactory factory = new DirectRabbitListenerContainerFactory();
-        factory.setConnectionFactory(connectionFactory);
-        connectionFactory.setExecutor(rabbitThreadPool);
-        return factory;
-    }
-
-}

src/main/java/com/mashibing/config/ThreadPoolConfig.java

@@ -11,6 +11,7 @@ import java.util.concurrent.TimeUnit;
 
 /**
  * 交给Hippo4j-Server管理的线程池
+ *
  * @author zjw
  * @description
  */
@@ -19,7 +20,7 @@ public class ThreadPoolConfig {
 
     @Bean
     @DynamicThreadPool
-    public ThreadPoolExecutor testThreadPool(){
+    public ThreadPoolExecutor testThreadPool() {
         //1、 采用线程池Builder去构建
         ThreadPoolExecutor testThreadPool = ThreadPoolBuilder.builder()
                 .corePoolSize(10)
@@ -34,4 +35,40 @@ public class ThreadPoolConfig {
                 .build();
         return testThreadPool;
     }
+
+    @Bean
+    @DynamicThreadPool
+    public ThreadPoolExecutor ioThreadPool() {
+        //1、 采用线程池Builder去构建
+        ThreadPoolExecutor testThreadPool = ThreadPoolBuilder.builder()
+                .corePoolSize(10)
+                .maximumPoolSize(10)
+                .keepAliveTime(10)
+                .timeUnit(TimeUnit.SECONDS)
+                .workQueue(BlockingQueueTypeEnum.RESIZABLE_LINKED_BLOCKING_QUEUE)
+                .threadFactory("io")
+                .rejected(new ThreadPoolExecutor.AbortPolicy())
+                .threadPoolId("io")
+                .dynamicPool()
+                .build();
+        return testThreadPool;
+    }
+
+    @Bean
+    @DynamicThreadPool
+    public ThreadPoolExecutor cpuThreadPool() {
+        //1、 采用线程池Builder去构建
+        ThreadPoolExecutor testThreadPool = ThreadPoolBuilder.builder()
+                .corePoolSize(10)
+                .maximumPoolSize(10)
+                .keepAliveTime(10)
+                .timeUnit(TimeUnit.SECONDS)
+                .workQueue(BlockingQueueTypeEnum.RESIZABLE_LINKED_BLOCKING_QUEUE)
+                .threadFactory("cpu")
+                .rejected(new ThreadPoolExecutor.AbortPolicy())
+                .threadPoolId("cpu")
+                .dynamicPool()
+                .build();
+        return testThreadPool;
+    }
 }

src/main/java/com/mashibing/controller/Hippo4jController.java

@@ -0,0 +1,32 @@
+package com.mashibing.controller;
+
+import com.mashibing.service.Hippo4jService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @author zjw
+ * @description
+ */
+@RestController
+public class Hippo4jController {
+
+
+    @Autowired
+    private Hippo4jService service;
+
+
+    @GetMapping("/cpu")
+    public String cpu() throws InterruptedException {
+        Long result = service.doSomeCPUThing();
+        return result + "";
+    }
+
+    @GetMapping("/io")
+    public String io() throws Exception {
+        String result = service.doIOSomeThing();
+        return result;
+    }
+
+}

src/main/java/com/mashibing/mq/TestListener.java

@@ -1,19 +1,16 @@
-package com.mashibing.mq;
+//package com.mashibing.mq;
-
+//
-import com.mashibing.controller.TestController;
+//import com.mashibing.controller.TestController;
-import org.springframework.amqp.rabbit.annotation.RabbitListener;
+//import org.springframework.amqp.rabbit.annotation.RabbitListener;
-import org.springframework.stereotype.Component;
+//import org.springframework.stereotype.Component;
-
+//
-/**
+//
- * @author zjw
+//@Component
- * @description
+//public class TestListener {
- */
+//
-@Component
+//    @RabbitListener(queues = "hippo4j",containerFactory = "defaultContainerFactory")
-public class TestListener {
+//    public void consume(String message){
-
+//        System.out.println(Thread.currentThread().getName() + "：消费消息 --> " + message);
-    @RabbitListener(queues = "hippo4j",containerFactory = "defaultContainerFactory")
+//    }
-    public void consume(String message){
+//
-        System.out.println(Thread.currentThread().getName() + "：消费消息 --> " + message);
+//}
-    }
-
-}

src/main/java/com/mashibing/service/Hippo4jService.java

@@ -0,0 +1,11 @@
+package com.mashibing.service;
+
+/**
+ * @author zjw
+ * @description
+ */
+public interface Hippo4jService {
+    String doIOSomeThing() throws Exception;
+
+    Long doSomeCPUThing() throws InterruptedException;
+}

src/main/java/com/mashibing/service/impl/Hippo4jServiceImpl.java

@@ -0,0 +1,89 @@
+package com.mashibing.service.impl;
+
+import com.mashibing.service.Hippo4jService;
+import lombok.SneakyThrows;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * @author zjw
+ * @description
+ */
+@Service
+public class Hippo4jServiceImpl implements Hippo4jService {
+
+    @Resource
+    private ThreadPoolExecutor ioThreadPool;
+
+    @Resource
+    private ThreadPoolExecutor cpuThreadPool;
+
+    @Override
+    public String doIOSomeThing() throws Exception{
+        CountDownLatch latch = new CountDownLatch(3);
+        Future<String> job1Result = ioThreadPool.submit(() -> {
+            String result1 = job1();
+            latch.countDown();
+            return result1;
+        });
+        Future<String> job2Result = ioThreadPool.submit(() -> {
+            String result2 = job2();
+            latch.countDown();
+            return result2;
+        });
+        Future<String> job3Result = ioThreadPool.submit(() -> {
+            String result3 = job3();
+            latch.countDown();
+            return result3;
+        });
+        latch.await();
+        return job1Result.get() + job1Result.get() + job1Result.get();
+    }
+    @SneakyThrows
+    private String job1() {
+        Thread.sleep(100);
+        return "RedisResult!";
+    }
+    @SneakyThrows
+    private String job2() {
+        Thread.sleep(200);
+        return "ServiceResult!";
+    }
+    @SneakyThrows
+    private String job3() {
+        Thread.sleep(200);
+        return "MySQLResult!";
+    }
+
+
+
+    @Override
+    public Long doSomeCPUThing() throws InterruptedException {
+        AtomicLong atomicLong = new AtomicLong(0);
+        CountDownLatch latch = new CountDownLatch(2);
+        cpuThreadPool.execute(() -> {
+            incr(atomicLong);
+            latch.countDown();
+        });
+        cpuThreadPool.execute(() -> {
+            incr(atomicLong);
+            latch.countDown();
+        });
+        latch.await();
+        return atomicLong.get();
+    }
+    private void incr(AtomicLong atomicLong){
+        for (int i = 0; i < 5000000; i++) {
+            atomicLong.incrementAndGet();
+        }
+    }
+
+
+
+
+}

src/main/resources/application-dev.yml

@@ -0,0 +1,16 @@
+spring:
+  application:
+    name: hippo4j-client
+  dynamic:
+    thread-pool:
+      server-addr: 192.168.11.88:6691
+      username: admin
+      password: 123456
+      namespace: mashibing   # 租户名称
+      item-id: ${spring.application.name}   # 项目名称，需要与与服务名称保持一致。
+  rabbitmq:
+    host: 192.168.11.88
+    port: 5672
+    username: guest
+    password: guest
+    virtual-host: /

src/main/resources/application-test.yml

@@ -0,0 +1,10 @@
+spring:
+  application:
+    name: hippo4j-client
+  dynamic:
+    thread-pool:
+      server-addr: localhost:6691
+      username: admin
+      password: 123456
+      namespace: mashibing   # 租户名称
+      item-id: ${spring.application.name}   # 项目名称，需要与与服务名称保持一致。

src/main/resources/application.yml

@@ -1,18 +1,3 @@
 spring:
   profiles:
     active: dev
-  application:
-    name: hippo4j-client
-  dynamic:
-    thread-pool:
-      server-addr: 192.168.11.88:6691
-      username: admin
-      password: 123456
-      namespace: mashibing   # 租户名称
-      item-id: ${spring.application.name}   # 项目名称，需要与与服务名称保持一致。
-  rabbitmq:
-    host: 192.168.11.88
-    port: 5672
-    username: guest
-    password: guest
-    virtual-host: /