public class Test {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newCachedThreadPool();
        Task task = new Task();
        Future<Integer> result = executor.submit(task);
        executor.shutdown();
         
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e1) {
            e1.printStackTrace();
        }
         
        System.out.println("主线程在执行任务");
         
        try {
            System.out.println("task运行结果"+result.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
         
        System.out.println("所有任务执行完毕");
    }
}
class Task implements Callable<Integer>{
    @Override
    public Integer call() throws Exception {
        System.out.println("子线程在进行计算");
        Thread.sleep(3000);
        int sum = 0;
        for(int i=0;i<100;i++)
            sum += i;
        return sum;
    }
}

public static void main(String[] args) {
        Callable<String> callable1=()->{
            Thread.sleep(2000);
            return Thread.currentThread().getName();
        };
        Callable<String> callable2=()->{
            Thread.sleep(3000);
            return Thread.currentThread().getName();
        };
        FutureTask<String> futureTask1 = new FutureTask<>(callable1);// 将Callable写的任务封装到一个由执行者调度的FutureTask对象
        FutureTask<String> futureTask2 = new FutureTask<>(callable2);

        ExecutorService executor = Executors.newFixedThreadPool(2);        // 创建线程池并返回ExecutorService实例
        executor.execute(futureTask1);  // 执行任务
        executor.execute(futureTask2);
        //同时开启了两个任务
        long startTime = System.currentTimeMillis();
        while (true) {
            try {
                if(futureTask1.isDone() && futureTask2.isDone()){//  两个任务都完成
                    System.out.println("Done");
                    executor.shutdown();                          // 关闭线程池和服务
                    return;
                }

                if(!futureTask1.isDone()){ // 任务1没有完成，会等待，直到任务完成
                    System.out.println("FutureTask1 output="+futureTask1.get());
                }

                System.out.println("Waiting for FutureTask2 to complete");
                String s = futureTask2.get(200L, TimeUnit.MILLISECONDS);
                if(s !=null){
                    System.out.println("FutureTask2 output="+s);
                }
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }catch(TimeoutException e){
                //do nothing
            }
            System.out.println((System.currentTimeMillis()-startTime));
        }
    }

public class Test {
    public static void main(String[] args) {
        //第一种方式
        ExecutorService executor = Executors.newCachedThreadPool();
        Task task = new Task();
        FutureTask<Integer> futureTask = new FutureTask<Integer>(task);
        executor.submit(futureTask);
        executor.shutdown();
         
        //第二种方式，注意这种方式和第一种方式效果是类似的，只不过一个使用的是ExecutorService，一个使用的是Thread
        /*Task task = new Task();
        FutureTask<Integer> futureTask = new FutureTask<Integer>(task);
        Thread thread = new Thread(futureTask);
        thread.start();*/
         
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e1) {
            e1.printStackTrace();
        }
         
        System.out.println("主线程在执行任务");
         
        try {
            System.out.println("task运行结果"+futureTask.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
         
        System.out.println("所有任务执行完毕");
    }
}
class Task implements Callable<Integer>{
    @Override
    public Integer call() throws Exception {
        System.out.println("子线程在进行计算");
        Thread.sleep(3000);
        int sum = 0;
        for(int i=0;i<100;i++)
            sum += i;
        return sum;
    }
}

package com.lhx.cloud.futruetask;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class CompletionServiceDemo {
    public static void main(String[] args)  {
        Long start = System.currentTimeMillis();
        //开启5个线程
        ExecutorService exs = Executors.newFixedThreadPool(5);
        try {
            int taskCount = 10;
            //结果集
            List<Integer> list = new ArrayList<>();
            //1.定义CompletionService
            CompletionService<Integer> completionService = new ExecutorCompletionService<>(exs);
            List<Future<Integer>> futureList = new ArrayList<>();
            //2.添加任务
            for(int i=0;i<taskCount;i++){
                futureList.add(completionService.submit(new Task(i+1)));
            }
            //==================结果归集===================
            //方法1：future是提交时返回的，遍历queue则按照任务提交顺序，获取结果
//            for (Future<Integer> future : futureList) {
//                System.out.println("====================");
//                Integer result = future.get();//线程在这里阻塞等待该任务执行完毕,按照
//                System.out.println("任务result="+result+"获取到结果!"+new Date());
//                list.add(result);
//            }

//            //方法2.使用内部阻塞队列的take()
            for(int i=0;i<taskCount;i++){
                Integer result = completionService.take().get();//采用completionService.take()，内部维护阻塞队列，任务先完成的先获取到
                System.out.println(LocalDateTime.now()+"---任务i=="+result+"完成!");
                list.add(result);
            }
            System.out.println("list="+list);
            System.out.println("总耗时="+(System.currentTimeMillis()-start));
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            exs.shutdown();//关闭线程池
        }

    }


    static class Task implements Callable<Integer>{
        Integer i;

        public Task(Integer i) {
            super();
            this.i=i;
        }

        @Override
        public Integer call() throws Exception {
            if(i==5){
                Thread.sleep(5000);
            }else{
                Thread.sleep(1000);
            }
            System.out.println("线程："+Thread.currentThread().getName()+"任务i="+i+",执行完成！");
            return i;
        }

    }
}

public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            //长时间的计算任务
            try {
                System.out.println("计算型任务开始");
                Thread.sleep(2000);
                return "计算型任务结束";
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "·00";
        });
        System.out.println(future.get());
    }

package com.lhx.cloud.futruetask;

import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

public class BasicFuture {

    private static Random rand = new Random();
    private static long t = System.currentTimeMillis();

    static int getMoreData()  {
        System.out.println("begin to start compute");
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("end to compute,passed " + (System.currentTimeMillis()-t));
        return rand.nextInt(1000);
    }


    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(BasicFuture::getMoreData);
        Future<Integer> f = future.whenComplete((v, e) -> {
            System.out.println(v);
            System.out.println(e);
        });
        System.out.println(f.get());
    }}

/**
     * 任务没有异常 正常执行，然后结束
     */
    @Test
    public void test1() throws ExecutionException, InterruptedException {
        CompletableFuture<String> completableFuture = new CompletableFuture<>();
        new Thread(() -> {
            // 模拟执行耗时任务
            System.out.println("task doing...");
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            // 告诉completableFuture任务已经完成
            completableFuture.complete("ok");
        }).start();
        // 获取任务结果，如果没有完成会一直阻塞等待
        String result = completableFuture.get();
        System.out.println("计算结果:" + result);
    }

    /**
     * 线程有异常  正常执行，然后无法结束，主线程会一直等待
     */
    @Test
    public void test2() throws ExecutionException, InterruptedException {
        CompletableFuture<String> completableFuture = new CompletableFuture<>();
        new Thread(() -> {
            // 模拟执行耗时任务
            System.out.println("task doing...");
            try {
                Thread.sleep(3000);
                int i=1/0;
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            // 告诉completableFuture任务已经完成
            completableFuture.complete("ok");
        }).start();
        // 获取任务结果，如果没有完成会一直阻塞等待
        String result = completableFuture.get();
        System.out.println("计算结果:" + result);
    }

package com.lhx.cloud.futruetask;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CompletableFutureDemo {
    public static void main(String[] args) {
        Long start = System.currentTimeMillis();
        // 结果集
        List<String> list = new ArrayList<>();

        ExecutorService executorService = Executors.newFixedThreadPool(10);

        List<Integer> taskList = Arrays.asList(2, 1, 3, 4, 5, 6, 7, 8, 9, 10);
        // 全流式处理转换成CompletableFuture[]+组装成一个无返回值CompletableFuture，join等待执行完毕。返回结果whenComplete获取
        CompletableFuture[] cfs = taskList.stream()
                .map(integer -> CompletableFuture.supplyAsync(() -> calc(integer), executorService)
                        .thenApply(h -> Integer.toString(h))
                        .whenComplete((s, e) -> {
                            System.out.println(LocalDateTime.now()+"---任务" + s + "完成!result=" + s + "，异常 e=" + e);
                            list.add(s);
                        })
                ).toArray(CompletableFuture[]::new);
        // 封装后无返回值，必须自己whenComplete()获取
        CompletableFuture.allOf(cfs).join();
        System.out.println("list=" + list + ",耗时=" + (System.currentTimeMillis() - start));
    }

    public static Integer calc(Integer i) {
        try {
            if (i == 1) {
                Thread.sleep(3000);//任务1耗时3秒
            } else if (i == 5) {
                Thread.sleep(5000);//任务5耗时5秒
            } else {
                Thread.sleep(1000);//其它任务耗时1秒
            }
            System.out.println(LocalDateTime.now()+"---task线程：" + Thread.currentThread().getName()
                    + "任务i=" + i + ",完成！" );
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return i;
    }
}

@Component
public class TaskFutureDemo {
    @Async
    public Future<String> doTaskOne() throws Exception {
        for (int i = 0; i < 3; i++) {
            Thread.sleep(1000);
            System.out.println("ThreadId:"+Thread.currentThread().getId()+":doTaskOne");
        }
        return new AsyncResult<>("doTaskOne");
    }

    @Async
    public Future<String> doTaskTwo() throws Exception {
        for (int i = 0; i < 3; i++) {
            Thread.sleep(1000);
            System.out.println("ThreadId:"+Thread.currentThread().getId()+":doTaskTwo");
        }
        return new AsyncResult<>("doTaskTwo");
    }

    @Async
    public Future<String> doTaskThree() throws Exception {
        for (int i = 0; i < 3; i++) {
            Thread.sleep(1000);
            System.out.println("ThreadId:"+Thread.currentThread().getId()+":doTaskThree");
        }
        return new AsyncResult<>("doTaskThree");
    }
}

@RestController
public class TestAsynsFutureController {
    @Autowired
    private TaskFutureDemo task;

    @RequestMapping("/testAsyncFuture")
    public ResponseEntity testAsyncFuture() throws Exception {
        Future<String> taskOne = task.doTaskOne();
        Future<String> taskTwo = task.doTaskTwo();
        Future<String> taskThree = task.doTaskThree();
        while (true) {
            if (taskOne.isDone() && taskTwo.isDone() && taskThree.isDone()) {
                break;
            }
        }
        return ResponseEntity.ok("ok");
    }
}

@Service
public class TaskListenableFutureService {
    private AsyncSearch asyncSearch;

    @Autowired
    public TaskListenableFutureService(AsyncSearch asyncSearch) {
        this.asyncSearch=asyncSearch;
    }

    public List<String> search(List<String> keywords){
        CountDownLatch latch=new CountDownLatch(keywords.size());
        List<String> allResult=Collections.synchronizedList(new ArrayList<>());
        keywords.stream()
                .forEach(keyword->asyncFetch(latch,allResult,keyword));
        await(latch);
        return allResult;
    }

    private void asyncFetch(CountDownLatch latch, List<String> result, String keyword){
        asyncSearch.asyncFetch(keyword)
                .addCallback(
                        key->onSuccess(result,latch,key),
                        ex -> onError(latch,ex)
                );
    }

    private void await(CountDownLatch latch){
        try {
            latch.await();
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        }
    }
    private static void onSuccess(List<String> result,CountDownLatch latch,String keyword){
        result.add(keyword);
        latch.countDown();
    }
    private static void onError(CountDownLatch latch,Throwable ex){
        ex.printStackTrace();
        latch.countDown();
    }
    @Component
    private static class AsyncSearch{
        @Autowired
        public AsyncSearch() {
        }

        protected final Logger logger = LoggerFactory.getLogger(AsyncSearch.class);
        @Async
        public ListenableFuture<String> asyncFetch(String keyword){
            logger.info(Thread.currentThread().getName()+"-"+keyword);
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return new AsyncResult<>("keyword="+keyword);
        }
    }
}

@RestController
public class TestAsynsListenableFutureController {
    protected final Logger logger = LoggerFactory.getLogger(TestAsynsListenableFutureController.class);
    @Autowired
    private TaskListenableFutureService task;

    @RequestMapping("/testAsyncListenableFuture")
    public ResponseEntity testAsyncListenableFuture() throws Exception {
        List<String> list = task.search(Arrays.asList("java", "html", "spring"));
        list.stream().forEach(p-> logger.info(p));

        return ResponseEntity.ok("ok");
    }
}

import com.fasterxml.jackson.annotation.JsonIgnoreProperties;

@JsonIgnoreProperties(ignoreUnknown=true)
public class User {

    private String name;
    private String blog;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getBlog() {
        return blog;
    }

    public void setBlog(String blog) {
        this.blog = blog;
    }

    @Override
    public String toString() {
        return "User [name=" + name + ", blog=" + blog + "]";
    }

}

@Service
public class GitHubLookupService {

    private static final Logger logger = LoggerFactory.getLogger(GitHubLookupService.class);

    private final RestTemplate restTemplate;

    public GitHubLookupService(RestTemplateBuilder restTemplateBuilder) {
        this.restTemplate = restTemplateBuilder.build();
    }

    @Async
    public CompletableFuture<User> findUser(String user) throws InterruptedException {
        logger.info("Looking up " + user);
        String url = String.format("https://api.github.com/users/%s", user);
        User results = restTemplate.getForObject(url, User.class);
        // Artificial delay of 1s for demonstration purposes
        Thread.sleep(1000L);
        return CompletableFuture.completedFuture(results);
    }

}

@RestController
public class CompletableFutureController {


    private static final Logger logger = LoggerFactory.getLogger(CompletableFutureController.class);
    @Autowired
    private  GitHubLookupService gitHubLookupService;

    @RequestMapping("/testCompletableFuture")
    public ResponseEntity testCompletableFuture() throws Exception {
        // Start the clock
        long start = System.currentTimeMillis();

        // Kick of multiple, asynchronous lookups
        CompletableFuture<User> page1 = gitHubLookupService.findUser("PivotalSoftware");
        CompletableFuture<User> page2 = gitHubLookupService.findUser("CloudFoundry");
        CompletableFuture<User> page3 = gitHubLookupService.findUser("Spring-Projects");

        // Wait until they are all done
        CompletableFuture.allOf(page1,page2,page3).join();

        // Print results, including elapsed time
        logger.info("Elapsed time: " + (System.currentTimeMillis() - start));
        logger.info("--> " + page1.get());
        logger.info("--> " + page2.get());
        logger.info("--> " + page3.get());

        return ResponseEntity.ok("ok");
    }
}

　　简而言之，WebSocket 是底层协议，SockJS 是WebSocket 的备选方案，也是 底层协议，而 STOMP 是基于 WebSocket（SockJS） 的上层协议

1. 假设HTTP协议并不存在，只能使用TCP套接字来编写web应用，你可能认为这是一件疯狂的事情。
2. 不过幸好，我们有HTTP协议，它解决了 web 浏览器发起请求以及 web 服务器响应请求的细节。
3. 直接使用 WebSocket（SockJS） 就很类似于 使用 TCP 套接字来编写 web 应用；因为没有高层协议，因此就需要我们定义应用间所发送消息的语义，还需要确保 连接的两端都能遵循这些语义。
4. 同HTTP在TCP套接字上添加请求-响应模型层一样，STOMP在 WebSocket之上提供了一个基于帧的线路格式层，用来定义消息语义。

• 实现 WebSocketConfigurer 接口，重写 registerWebSocketHandlers 方法，这是一个核心实现方法，配置 websocket 入口，允许访问的域、注册 Handler、SockJs 支持和拦截器。
• registry.addHandler()注册和路由的功能，当客户端发起 websocket 连接，把 /path 交给对应的 handler 处理，而不实现具体的业务逻辑，可以理解为收集和任务分发中心。
• addInterceptors，顾名思义就是为 handler 添加拦截器，可以在调用 handler 前后加入我们自己的逻辑代码。
• ServletServerContainerFactoryBean可以添加对WebSocket的一些配置

对以上代码分析：

1. EnableWebSocketMessageBroker 注解表明： 这个配置类不仅配置了 WebSocket，还配置了基于代理的 STOMP 消息；
2. 它复写了 registerStompEndpoints() 方法：添加一个服务端点，来接收客户端的连接。将 “/socket-server-point” 路径注册为 STOMP 端点。这个路径与之前发送和接收消息的目的路径有所不同， 这是一个端点，客户端在订阅或发布消息到目的地址前，要连接该端点，即用户发送请求 ：url=’/127.0.0.1:8080/socket-server-point’ 与 STOMP server 进行连接，之后再转发到订阅url；
3. 它复写了 configureMessageBroker() 方法：配置了一个 简单的消息代理，通俗一点讲就是设置消息连接请求的各种规范信息。
4. 发送应用程序的消息将会带有 “/ws” 前缀。

@Service
public class HandlerService {
    private static final Logger logger = LoggerFactory.getLogger(HandlerService.class);
    @Async
    public CompletableFuture<String> handle(String key) throws Exception {
        Thread.sleep(new Random().nextInt(3000));
        logger.info("Looking up " + key);
        key=key+":"+LocalDateTime.now();
        return CompletableFuture.completedFuture(key);
    }
}

@MessageMapping("/searchBase")
    public ResponseEntity searchBase() throws Exception {
        Consumer<List<String>> callback = p -> websocket.convertAndSend("/topic/searchResults", p);
        List<String> list = Arrays.asList("bba", "aaa", "ccc");
        localSearch(list, callback);
        Map map = new HashMap();
        map.put("list", list);
        map.put("date", LocalDateTime.now());
        return ResponseEntity.ok(map);
    }

    public void localSearch(List<String> keys, Consumer<List<String>> callback) throws Exception {
        Thread.sleep(2000);
        List<String> list = new ArrayList<>();
        for (String key : keys) {
            CompletableFuture<String> completableFuture = handlerService.handle(key);
            completableFuture.thenAcceptAsync(p -> {
                list.clear();
                list.add(p);
                callback.accept(list);
            });
        }
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableStompBrokerRelay("/queue", "/topic");
        //定义了一个客户端订阅地址的前缀信息，也就是客户端接收服务端发送消息的前缀信息
//        registry.enableSimpleBroker("/topic");
        //定义了服务端接收地址的前缀，也即客户端给服务端发消息的地址前缀
        registry.setApplicationDestinationPrefixes("/ws");
    }

  @Override
  public void configureMessageBroker(MessageBrokerRegistry registry) {
    registry.enableStompBrokerRelay("/queue", "/topic")
            .setRelayHost("rabbit.someotherserver")
            .setRelayPort(62623)
            .setClientLogin("marcopolo")
            .setClientPasscode("letmein01")
    registry.setApplicationDestinationPrefixes("/app");
  }

/**
     * 处理来自客户端的STOMP消息
     * @param incoming
     * @return
     */
    @MessageMapping("/incoming")
    public Shout handleShout(Shout incoming) {
        logger.info("Received message: " + incoming.getMessage());
        try { Thread.sleep(2000); } catch (InterruptedException e) {}
        Shout outgoing = new Shout();
        outgoing.setMessage("incoming!");
        return outgoing;
    }

public class Shout {
    private String message;

    public String getMessage() {
        return message;
    }

    public void setMessage(String message) {
        this.message = message;
    }
}

function contect() {
    var socket=new SockJS(''/socket-server-point'');
    stompCliet=Stomp.over(socket);
    stompCliet.connect({},function (frame) {
        console.log(''Connected ：''+frame);
        stompCliet.send("/ws/incoming",{},"{\"message\":\"Hello!\"}");
    });
}
contect();

@SubscribeMapping("/sub")
    public Shout handleSubscription(){
        logger.info("Received message: " +"subscription");
        Shout outgoing = new Shout();
        outgoing.setMessage("subscription!");
        return outgoing;
    }

function contect() {
    var socket=new SockJS(''/socket-server-point'');
    stompCliet=Stomp.over(socket);
    stompCliet.connect({},function (frame) {
        console.log(''Connected ：''+frame);
        stompCliet.subscribe(''/ws/sub'',function (result) {
            console.log("aaaa",JSON.parse(result.body));
        });
        stompCliet.send("/ws/sub",{},"{\"message\":\"Hello!\"}");
    });
}
contect();

@MessageMapping("/incoming")
    public Shout handleShout(Shout incoming) {
        logger.info("Received message: " + incoming.getMessage());
        try { Thread.sleep(2000); } catch (InterruptedException e) {}
        Shout outgoing = new Shout();
        outgoing.setMessage("incoming!");
        return outgoing;
    }

@MessageMapping("/incoming")
    @SendTo("/topic/shout")
    public Shout handleShout(Shout incoming) {
        logger.info("Received message: " + incoming.getMessage());
        try { Thread.sleep(2000); } catch (InterruptedException e) {}
        Shout outgoing = new Shout();
        outgoing.setMessage("incoming!");
        return outgoing;
    }

@SubscribeMapping("/sub")
    public Shout handleSubscription(){
        logger.info("Received message: " +"subscription");
        Shout outgoing = new Shout();
        outgoing.setMessage("subscription!");
        return outgoing;
    }

stompCliet.subscribe(''/ws/sub'',function (result) {
            console.log("aaaa",JSON.parse(result.body));
        });

function contect() {
    var socket=new SockJS(''/socket-server-point'');
    stompCliet=Stomp.over(socket);
    stompCliet.connect({},function (frame) {
        console.log(''Connected ：''+frame);
        stompCliet.subscribe(''/topic/sendDataToClient'',function (result) {
            console.log("aaaa",JSON.parse(result.body));
        });
        // stompCliet.send("/ws/sub",{},"{\"message\":\"Hello!\"}");
    });
}
contect();

@Autowired(required = false)
    private SimpMessagingTemplate websocket;

    @GetMapping("/sendDataToClient")
    public ResponseEntity sendDataToClient() throws Exception {
        Map map=new HashMap();
        map.put("aa","aaa");
        map.put("bb","bbb");
        websocket.convertAndSend("/topic/sendDataToClient",map);
        return ResponseEntity.ok("ok");
    }

@MessageMapping("/shout")
    @SendToUser("/queue/notifications")
    public Shout userStomp(Principal principal, Shout shout) {
        String name = principal.getName();
        String message = shout.getMessage();
        logger.info("认证的名字是：{}，收到的消息是：{}", name, message);
        return shout;
    }

var headers={
    username:''admin'',
    password:''admin''
};
function contect() {
    var socket=new SockJS(''/socket-server-point'');
    stompCliet=Stomp.over(socket);
    stompCliet.connect(headers,function (frame) {
        console.log(''Connected ：''+frame);
        stompCliet.subscribe(''/user/queue/notifications'',function (result) {
            console.log("aaaa",JSON.parse(result.body));
        });
        stompCliet.send("/ws/shout",{},"{\"message\":\"Hello!\"}");
    });
}
contect();

@MessageMapping("/singleShout")
    public void singleUser(Shout shout, StompHeaderAccessor stompHeaderAccessor) {
        String message = shout.getMessage();
        LOGGER.info("接收到消息：" + message);
        Principal user = stompHeaderAccessor.getUser();
        simpMessageSendingOperations.convertAndSendToUser(user.getName(), "/queue/shouts", shout);
    }

@MessageExceptionHandler(Exception.class)
     @SendToUser("/queue/errors")
     public Exception handleExceptions(Exception t){
         t.printStackTrace();
         return t;
     }

//默认的和STOMP端点连接
/*stomp.connect("guest", "guest", function (franme) {
});*/

var headers={
    username:''admin'',
    password:''admin''
};

stomp.connect(headers, function (frame) {

// 建立连接对象（还未发起连接）
 var socket=new SockJS("/endpointChat"); 
 // 获取 STOMP 子协议的客户端对象 
 var stompClient = Stomp.over(socket); 
 // 向服务器发起websocket连接并发送CONNECT帧 
 stompClient.connect( {}, 
function connectCallback (frame) { 
     // 连接成功时（服务器响应 CONNECTED 帧）的回调方法 
console.log(''已连接【'' + frame + ''】''); 
//订阅一个消息
     stompClient.subscribe(''/topic/getResponse'',
function (response) { 
showResponse(response.body);
});
 },
     function errorCallBack (error) { 
     // 连接失败时（服务器响应 ERROR 帧）的回调方法 
console.log(''连接失败【'' + error + ''】''); 
} );

client.disconnect(
function () { 
    alert("断开连接");
});

client.send(destination url, headers, body);

client.send("/queue/test", {priority: 9}, "Hello, STOMP");
client.send("/queue/test", {}, "Hello, STOMP");

subscribe(destination url, callback, headers)

var headers = {
ack: ''client'',
//这个客户端指定了它会确认接收的信息，只接收符合这个selector : location = ''Europe''的消息。
 ''selector'': "location = ''Europe''",
//id:’myid’
}; 
var callback = function(message) {
if (message.body) {
 alert("got message with body " +JSON.parse( message.body)) }
 else{
alert("got empty message"); 
} }); 
var subscription = client.subscribe("/queue/test", callback, headers);
 
如果想让客户端订阅多个目的地，你可以在接收所有信息的时候调用相同的回调函数：
onmessage = function(message) {
    // called every time the client receives a message
}
var sub1 = client.subscribe("queue/test", onmessage);
var sub2 = client.subscribe("queue/another", onmessage)

var subscription = client.subscribe(...);
 
subscription.unsubscribe();

commit()提交事务
 
abort()中止事务

// start the transaction
 
var tx = client.begin();
 
// send the message in a transaction
 
client.send("/queue/test", {transaction: tx.id}, "message in a transaction");
 
// commit the transaction to effectively send the message
 
tx.commit();

var txid = "unique_transaction_identifier";
 
// start the transaction
 
var tx = client.begin();
 
// oops! send the message outside the transaction
 
client.send("/queue/test", {}, "I thought I was in a transaction!");
 
tx.abort(); // Too late! the message has been sent

var subscription = client.subscribe("/queue/test",
    function(message) {
        // do something with the message
        ...
        // and acknowledge it
        message.ack();
    },
    {ack: ''client''}
);

var tx = client.begin();
message.ack({ transaction: tx.id, receipt: ''my-receipt'' });
tx.commit();

client.debug = function(str) {
 
    // append the debug log to a #debug div somewhere in the page using JQuery:
 
    $("#debug").append(str + "\n");
};

client.heartbeat.outgoing = 20000;
 
// client will send heartbeats every 20000ms
 
client.heartbeat.incoming = 0;
 
// client does not want to receive heartbeats
 
// from the server