push-server

KO | EN

gitlite — search

#react #typescript #go #android #rust #deep-learning #nodejs #macos #java #windows #cli #vue #security #open-source

(★ 32)

主要是基于mqtt的消息推送系统，单点推送，分组推送

push-server 최신버젼 다운로드

최종 버전 다운로드 (.zip)

// repository documentation

Was this content helpful?

★ 0 (0 ratings)

1、主要是基于mqtt的消息推送服务，本系统的设计采用springboot+mqtt来实现消息推送
2、采用kafka作为数据获取后暂存点，可以通过接口获取数据（sub），或则通过客户端发送到mqtt订阅的方式获取对应的消息
3、消息传输格式采用json，内部定义了一些常用的消息作为本服务的一个约束
4、需要搭建mqtt服务器，集群或则单机，具体搭建需要参考mqtt
5、这边项目还需要redis作为缓存工具，主要是缓存用户的发送信息

@RestController
@RequestMapping("/api/push-server/")
public class RegisterController {
    @Autowired
    private RedisTemplate redisTemplate;

/**
     * 进行注册
     */
    @RequestMapping(value = "unRegist",method = RequestMethod.POST)
    public void unRegist(RegistDto registDto){
        redisTemplate.boundSetOps(registDto.getTopicPre()).add(registDto.getMark());
    }

/**
     * 取消注册
     * @param registDto
     */
    @RequestMapping(value = "regist",method = RequestMethod.POST)
    public void regist(RegistDto registDto){
        redisTemplate.boundSetOps(registDto.getTopicPre()).remove(registDto.getMark());
    }
}
进行对应的订阅渠道注册，主要是消息推送的时候能针对性的推送

/**
 * @Auther yidadi
 * @Date 17-9-1 下午4:31
 */
@RestController
@RequestMapping("/api/push-server/")
public class PushController {
    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private MqttClient mqttProducer;

/**
     *
     * @param pushMsgDto
     */
    @RequestMapping(value = "pushTopic",method = RequestMethod.POST)
    public void pushTopic(PushMsgDto pushMsgDto){
        Set<String> datas = redisTemplate.boundSetOps(pushMsgDto.getTopicPre()).members();
        if(datas.isEmpty()){
            throw new PushException("the register user is empty");
        }else{
            datas.forEach(data->{
                MqttMessage message = new MqttMessage();
                message.setQos(pushMsgDto.getQos());
                message.setRetained(false);
                message.setPayload(pushMsgDto.getContent().getBytes());
                try {
                    mqttProducer.publish(pushMsgDto.getTopicPre().concat(data),message);
                } catch (MqttException e) {
                    e.printStackTrace();
                }
            });
        }
    }

/**
     *
     * @param pushMsgDto
     */
    @RequestMapping(value = "pushOne",method = RequestMethod.POST)
    public void pushOne(PushMsgDto pushMsgDto){
        MqttMessage message = new MqttMessage();
        message.setQos(pushMsgDto.getQos());
        message.setRetained(false);
        message.setPayload(pushMsgDto.getContent().getBytes());
        try {
            mqttProducer.publish(pushMsgDto.getTopicPre().concat(pushMsgDto.getMark()),message);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}

可以直接调用接口的方式发送推送消息，可以单点推送也可针对对应的渠道推送

/**
 * @Auther yidadi
 * @Date 17-9-4 下午3:42
 */
@RestController
@RequestMapping("/api/push-server/")
public class SubController {
    @Autowired
    private KafkaProducer kafkaProducer;

/**
     *
     * @param subMsgDto
     */
    @RequestMapping(value = "sub",method = RequestMethod.POST)
    public void sub(SubMsgDto subMsgDto){
        ProducerRecord producerRecord = new ProducerRecord(subMsgDto.getMqtopic(), JsonUtils.toJson(subMsgDto));
        kafkaProducer.send(producerRecord);
    }
} 
可以通过接口方式的获取客户端发送回来的数据

/**
 * @Auther yidadi
 * @Date 17-9-4 下午3:38
 */
public class MqttCallBack implements MqttCallback {
    private KafkaProducer kafkaProducer;

public MqttCallBack(KafkaProducer kafkaProducer) {
        this.kafkaProducer = kafkaProducer;
    }

public void connectionLost(Throwable cause) {
        cause.printStackTrace();
    }

@Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        SubMsgDto subMsgDto = (SubMsgDto)JsonUtils.readFromJson(new String(message.getPayload()),SubMsgDto.class);
        kafkaProducer.send(new ProducerRecord(subMsgDto.getMqtopic(),JsonUtils.toJson(subMsgDto)));
    }

@Override
    public void deliveryComplete(IMqttDeliveryToken token) {

}
}

也可以通过mqtt的方式获取对应的渠道消息发送到kafka，消息体里面需要制定对应的kafka的topic

push-server

push-server 최신버젼 다운로드

오늘의 추천 프로젝트

Was this content helpful?

Recent Feedback

push-server

push-server 최신버젼 다운로드

오늘의 추천 프로젝트

Was this content helpful?

Recent Feedback

Download README